mirror of https://github.com/apache/activemq.git
If multiple concurrent threads were creating vm://localhost connection, it was possible multiple "localhost" broker would be started.
git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@388082 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
47f2bb7cc2
commit
61fef743e8
|
@ -16,7 +16,7 @@
|
||||||
*/
|
*/
|
||||||
package org.apache.activemq.broker;
|
package org.apache.activemq.broker;
|
||||||
|
|
||||||
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
|
import java.util.HashMap;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
*
|
*
|
||||||
|
@ -30,21 +30,29 @@ public class BrokerRegistry {
|
||||||
return instance;
|
return instance;
|
||||||
}
|
}
|
||||||
|
|
||||||
ConcurrentHashMap brokers = new ConcurrentHashMap();
|
private final Object mutex = new Object();
|
||||||
|
private final HashMap brokers = new HashMap();
|
||||||
private BrokerRegistry() {
|
|
||||||
}
|
|
||||||
|
|
||||||
public BrokerService lookup(String brokerName) {
|
public BrokerService lookup(String brokerName) {
|
||||||
|
synchronized(mutex) {
|
||||||
return (BrokerService)brokers.get(brokerName);
|
return (BrokerService)brokers.get(brokerName);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public void bind(String brokerName, BrokerService broker) {
|
public void bind(String brokerName, BrokerService broker) {
|
||||||
|
synchronized(mutex) {
|
||||||
brokers.put(brokerName, broker);
|
brokers.put(brokerName, broker);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public void unbind(String brokerName) {
|
public void unbind(String brokerName) {
|
||||||
|
synchronized(mutex) {
|
||||||
brokers.remove(brokerName);
|
brokers.remove(brokerName);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public Object getRegistryMutext() {
|
||||||
|
return mutex;
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -86,7 +86,12 @@ public class VMTransportFactory extends TransportFactory{
|
||||||
VMTransportServer server=(VMTransportServer) servers.get(host);
|
VMTransportServer server=(VMTransportServer) servers.get(host);
|
||||||
// validate the broker is still active
|
// validate the broker is still active
|
||||||
if(!validateBroker(host)||server==null){
|
if(!validateBroker(host)||server==null){
|
||||||
BrokerService broker=BrokerRegistry.getInstance().lookup(host);
|
BrokerService broker=null;
|
||||||
|
// Synchronize on the registry so that multiple concurrent threads
|
||||||
|
// doing this do not think that the broker has not been created and cause multiple
|
||||||
|
// brokers to be started.
|
||||||
|
synchronized( BrokerRegistry.getInstance().getRegistryMutext() ) {
|
||||||
|
broker=BrokerRegistry.getInstance().lookup(host);
|
||||||
if(broker==null){
|
if(broker==null){
|
||||||
try{
|
try{
|
||||||
if(brokerFactoryHandler!=null){
|
if(brokerFactoryHandler!=null){
|
||||||
|
@ -100,6 +105,7 @@ public class VMTransportFactory extends TransportFactory{
|
||||||
}
|
}
|
||||||
brokers.put(host,broker);
|
brokers.put(host,broker);
|
||||||
}
|
}
|
||||||
|
|
||||||
server=(VMTransportServer) servers.get(host);
|
server=(VMTransportServer) servers.get(host);
|
||||||
if(server==null){
|
if(server==null){
|
||||||
server=(VMTransportServer) bind(location,true);
|
server=(VMTransportServer) bind(location,true);
|
||||||
|
@ -108,7 +114,10 @@ public class VMTransportFactory extends TransportFactory{
|
||||||
connector.start();
|
connector.start();
|
||||||
connectors.put(host,connector);
|
connectors.put(host,connector);
|
||||||
}
|
}
|
||||||
}else{}
|
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
VMTransport vmtransport=server.connect();
|
VMTransport vmtransport=server.connect();
|
||||||
IntrospectionSupport.setProperties(vmtransport,options);
|
IntrospectionSupport.setProperties(vmtransport,options);
|
||||||
Transport transport=vmtransport;
|
Transport transport=vmtransport;
|
||||||
|
|
Loading…
Reference in New Issue