solospider

2007-12-04

The patch of the open bug [AMQ-1116]

关键字: ActiveMQ

项目中使用Activemq作为消息中间件，ActiveMQ是Apache的开源项目，是一个轻量级的JMS实现。
在使用的过程中遇到了一个比较棘手的问题：
如果连接到ServiceBroker的Connection设成failover的话，当连接断开的时候，FailoverTransport会启动一个工作线程不断的重连，这个时候如果有一个消息需要发送的话，这个消息会被block住，直到重新建立连接才能发送。不过这个时候用户有可能会想要关闭这个Client端。这时候问题出现了，客户端无法关闭这个重连的线程。
原因如下：
ActiveMQ在通信层使用了管道过滤器的模式，当客户端要发送一个消息的时候，这个消息会经过一个Filter Chain进行层层处理，org.apache.activemq.transport.MutexTransport是其中的一个，主要作用是对线程加上互斥锁，代码如下：

    public void oneway(Object command) throws IOException {
        synchronized (writeMutex) {
            next.oneway(command);
        }
    }

在MutexTransport后面有一个FailoverTransport，担负着如果连接断开则重新连接的责任，所以FailoverTransport在网络断开的时候会阻塞住，这样writeMutex就无法释放，而用户的关闭请求也是作为一个command传入的，所以Shutdown这个Command就无法传入FailoverTransport通知其停止重连。
解决方案：
大概的思路就是在MutexTransport做一个判断，如果Command为ShutdownCommand就无须获取writeMutex锁，直接通过MutexTransport进入FailoverTransport,通知其停止重连。
MutexTransport的修改代码如下：

    
public void oneway(Object command) throws IOException {
         if (command instanceof ShutdownCommand) {
		next.oneway(command);
	} else {
                synchronized (writeMutex) {
                      next.oneway(command);
        	}
	}
}