回来看了一下java如果要实现这些功能，代码可能比shell还要多，看来这个领域实在不是java的专长。
我以前也认为语法甜点确认只是锦上添花，但是使用了python之后，发现自己以前还是偏见啊，在特定环境下，如脚本、页面等情况下，语法甜点可以大大减少输入量和代码出错可能性，动态语言还是我们工具箱不可缺少的工具。

其中，红色部分是系统（也就是启动java程序加载Main函数的classloader）,主要的设计考量有以下几点：

1、使用自定义的ExtClassLoader（加载java的ext目录下的jar包）把程序加载的class完全和系统加载的class隔离开，这样即使在eclipse容器中启动都不会有类冲突。

为什么不从系统的ExtClassLoader作为自定义classloader数的根有两个考虑，第一个是系统ExtClassLoader有可能不存在，第二个就是如果使用同一个ExtClassLoader中，在处理JNDI、XML和URL解析等java扩展功能时会遇到后加载的handler部分导致不同classloader树加载的同一个类的ClassCastException，具体参见这些模块的源代码。

2、WarClassLoader除了系统类和Common类（目前只有log相关类）以外的类都从war包的WEB-INFO和classes下加载。

3、所有执行War包中代码的线程ThreadContextClassLoader都设置为WarClassLoader，以供Spring和Webx中的相关工具类使用这个classloader结构的后门来加载war包中的类，典型例子是Webx中ResourceLoaderService就是使用ContextClassLoader来加载类的。

4、RialtoClassLoader也就是这个项目的容器加载器和WarClassLoader不在同一个树路径上，可以避免程序使用类和war使用类的class冲突，典型的是Spring容器相关代码。

5、CommonClassLoader加载的类需要严格控制，否则可能会导致运行期类冲突，例如Spring的相关jar包绝对不可以出现在这个classloader作用范围内。

总之，ClassLoader采用父分派机制，后来增加的Thread ContextClassLoader在这个体系上增加了一个后门，带来了灵活性，也带来了很多令人困扰的问题，在做容器类的项目时难免会遇到class loader层次设计的问题，这里抛砖引玉，欢迎达人拍砖。

在ActiveMQ中，Broker代表一个运行的MQ节点，ActiveMQ的插件实际上是基于Broker的一个Filter链，整个设计类似于servlet的Filter结构，所有的Plugin构成一个链式结构，每个插件实际上都是一个"Interceptor"，类结构图如下：

其中Broker接口封装了一个AMQ节点的方方面面的方法，包括连接管理、session管理、消息的发送和接收以及其它的一些功能，BrokerFilter实现这个接口，并提供了链式结构支持，可以拦截所有Broker方法的实现并传递结果给链式结构的下一个，形成了一个完整的"职责链"模式，具体层次关系如下，其中，"System Plugin"是指AMQ内部使用Plugin机制实现的一些系统功能，用户不能定制，"AMQ Plugin"指的是ActiveMQ已经实现好了，可以在配置文件中自由选择的一些插件，例如简单的安全插件，JAAS安全插件和DLQ插件等等，用户插件就是指用户自己实现的amq插件，需要用户把相关jar包放入到amq的启动classpath中，并在配置文件中进行配置才能正确加载的插件。

在上面这个层次结构中，最下面的RegionBroker是核心组件，在其之上的都是Broker的插件，继承之于BrokerFilter，和Broker保持接口兼容但是扩展Broker的功能。

下面举一个简单的例子，具体说明一下AMQ的插件是如何工作的。

我们在使用AMQ的过程中发现，在测试环境维护方面有很大的麻烦，具体表现在很多同学在测试项目的时候往往只关注自己项目牵涉的队列，不会去消费其他"不相关"的队列，这样导致的一个问题就是ActiveMQ经常发生大量数据阻塞，导致测试环境不可用，影响相关项目的测试工作。为了避免这个问题，我们假定在测试环境可以定义以下一些限制条件：

1、 所有队列堆积消息不超过1000条，超过之后立即清除。

2、 消息超过1个小时没有消费，就直接过期。

我们可以编写一个简单的amq插件来完成这两个限制条件：

首先，编写一个插件安装类：

package com.alibaba.napoli.plugins;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerPlugin;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class MessageControlBrokerPlugin implements BrokerPlugin {
private static Log log = LogFactory.getLog(StatisticsBrokerPlugin.class);

public Broker installPlugin(Broker broker) throws Exception {
log.info("install MessageControlBrokerPlugin");
return new MessageControlBroker(broker);
}
}

其次，编写真正的插件实现：

package com.alibaba.napoli.plugins;

import java.io.IOException;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.command.Message;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
* 开发环境管理插件，符合两个条件进行消息清理：<br>
* 1 消息累积超过1000条
* 2 消息超过1个小时无人消费
* @author guolin.zhuanggl
*
*/
public class MessageControlBroker extends BrokerFilter {
public static Log log = LogFactory.getLog(DiscardingDLQBroker.class);
private static final long DEFAULT_EXPIRATION = 3600*1000;
private static final long DEFAULT_PURGE_COUNT = 1000;

public MessageControlBroker(Broker next) {
super(next);
}

@Override
public void messageExpired(ConnectionContext context,
MessageReference message) {

Message msg = null;
try {
msg = message.getMessage();
} catch (IOException e) {
log.error("failed to fetch content: ",e);
}
purgeMessage(msg);
// TODO Auto-generated method stub
super.messageExpired(context, message);
}

/**
* 清除队列中的所有消息
*/
private void purgeMessage(Message message){
Destination r = message.getRegionDestination();
if(r instanceof Queue){
try {
//如果累积消息超过1000个，清除队列消息
if(((Queue) r).getMessages().size() > DEFAULT_PURGE_COUNT){
((Queue) r).purge();
}
} catch (Exception e) {
// TODO Auto-generated catch block
log.error("failed to purge queue "+r.getName(),e);
}
}

}
/**
* 当消息发送时，全部设置过期时间1个小时，测试环境专用！！！
*/
@Override
public void send(ProducerBrokerExchange producerExchange,Message messageSend) throws Exception {
long oldExp = messageSend.getExpiration();
messageSend.setExpiration(oldExp < DEFAULT_EXPIRATION && oldExp > 0 ? oldExp : DEFAULT_EXPIRATION );
purgeMessage(messageSend);
super.send(producerExchange, messageSend);
}

}

然后，将这两个类打包为myplugin.jar，并放在activemq启动目录下的lib目录下

最后，在activemq.xml文件中增加一个简单的spring配置项：

<bean xmlns="http://www.springframework.org/schema/beans"
id="purgePlugin"
class="com.alibaba.napoli.plugins.MessageControlBrokerPlugin">
</bean>

然后，重启activemq，就会发现这个插件已经被加载。

【背景】

在中文站 napoli 上线过程后，发现了一个奇怪的现象，尽管"已知"的 offer 发送端都已经迁移到
napoli 系统中，但是老的 mq 系统仍然有新的 offer 消息进来，因为连接 mq 的服务器非常多，定位消息来源成了一个非常大的问题。这种情况，想到了使用 BTrace 在某一台服务器进行线上监控进而期望发现这个幽灵。

【过程】

首先，我们需要知道两个基本信息：消息类型和来源 ip ，这样才可以定位 offer 消息的来源。

要知道来源 ip ，需要找到服务器端 管理的类，只有在建立 socket 的地方，才可以抓到具体 ip ，经过分析 amq 代码，发现 tcp 连接基本是由下面这个类来服务所有消息的接收的：

这个类中包含一个 socket 对象的成员变量，所有我们只要监控 readCommand 方法，这个方法的返回值实际上就是一个 ActivemqObjectMessage 对象，这样就可以在一个方法上加拦截器就可以同时捕获到 ip 和消息对象，两全其美！！！

因为原有 ESB 消息通道都是一个队列 ESBQueue ，所以无法通过队列名称来确定消息类型，必须通过
ESBTransferObject 对象来取得消息类型： destType ， offer 的区间是 1000-1008

但是，在服务器端并没有 ESBTransferObject 对象，无法反序列化（ BTrace 也不支持反序列化操作），所以没有方法简单取得消息类型信息！！！

OK ，我不反序列化，直接拿二进制 byte[] ，类型信息应该是在固定位置的吧？但是发现这个对象
content 变长字符串定义在类型之前，类型位置不确定了，晕倒啊 不死心，输出二进制数据，柳暗花明啊，原来对象序列化的时候， primitive 的 field 都是紧接着类型信息写入的，所以，类型信息是在固定位置的 ，类型信息始终是 255 ， 256 两个字节（实际上是 4 个字节，但是目前我们只占有 2 个） Ok ，编写代码，测试环境运行一下，晕倒，竟然有数组溢出！

使用 BTrace ，把这个数组打印下来（这个需要点技巧， btrace 连 for 都不允许），竟然发现 位置偏移到 205 ， 206 位置 ，这个真的不知道什么原因，估计是客户端发送的时候压缩了，简单修改偏移量，测试运行， ok ，所有的消息类型和 ip 的对照表打印出来了。

package com.alibaba.btrace.script;

import static com.sun.btrace.BTraceUtils.*;

import com.sun.btrace.annotations.*;

@BTrace

public class AMQQueue2IP {

@OnMethod(clazz = "org.apache.activemq.transport.tcp.TcpTransport", //需要拦截的类名

method = "readCommand", //需要拦截的方法名

location = @Location(Kind.RETURN)) //拦截位置，方法返回时

public static void onTransportCommandExit(@Self Object transport, @Return Object command) { //捕获调用对象和返回值

String commandName = str(command);

boolean isObjectMessage = (indexOf(commandName, "org.apache.activemq.command.ActiveMQObjectMessage") >= 0);

if (isObjectMessage) {

Object msg = command;

Object content = get(field(getSuperclass(getSuperclass(classOf(msg))), "content", false), msg);//捕获消息内容byte[]

byte[] bs = (byte[]) get(field(classOf(content), "data", false), content);

if (bs.length >= 206) {

int off = getInt(field(classOf(content), "offset", false), content);

int code = (0xff00&bs[205]<<8)+(0xff&bs[206]); //转换205,206字节为消息类型

//println(str(code));

Object socket = get(field(classOf(transport), "socket"), transport);

String address = str(socket); //截取ip地址

int s = indexOf(address, "/");

int e = indexOf(address, ",");

int len = e - s;

String ip = substr(address, s + 1, e);

print(strcat(timestamp(),"---"));

println(strcat(strcat("ip: ", ip), strcat(" queueName: ", str(code))));

}

}

}

}

打印结果：

2/3/10 12:38 PM---ip: 172.22.2.34 queueName: 2001

2/3/10 12:38 PM---ip: 172.22.2.41 queueName: 5001

2/3/10 12:38 PM---ip: 172.22.2.22 queueName: 5001

2/3/10 12:38 PM---ip: 172.22.2.47 queueName: 2001

2/3/10 12:38 PM---ip: 172.22.2.31 queueName: 2001

2/3/10 12:38 PM---ip: 172.22.2.13 queueName: 5001

2/3/10 12:38 PM---ip: 172.22.2.6 queueName: 5001

2/3/10 12:38 PM---ip: 172.22.2.48 queueName: 2001

2/3/10 12:38 PM---ip: 172.22.2.39 queueName: 2001

【补充】

BTrace 是一个强大的工具，但是，在线上检测的时候考虑时效性和安全性，必须有一个经过检验的脚本库才可以安全及时的定位系统问题.

但是，为什么文本文件会被当成是二进制文件？和今天的DNS切换有什么关系？分析后发现，因为dns的问题，抓数据的脚本执行时间明显变长，这样，在文件还在写入的时候，监控脚本就开始读取数据文件，在这样的并发访问下，grep会认为自己正在访问一个binary文件，导致监控误报警。

调试代码发现：

在Activemq的send response处理中，使用了一个BlockingQueue，在有timeout的方法里，使用了poll方法，这个方法的api说明中指出，当timeout发生时，这个方法返回null！！！

我们在看AMQ经过层层调用后，在ActiveMQConnection方法中如何处理这个返回值：

对返回值为空的情况没有做任何处理，即使消息发送超时，amq也认为这个消息发送成功！估计这哥们理解poll在timeout的时候会抛出异常吧。

解决办法很简单，在response为空的时候，抛出JMSException，告知发生Timeout错误。