本文主要是介绍ActiveMQ源码架构解析第二节，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
ActiveMQ源码架构解析第二节
- 博客分类：
- ActiveMQ
本节主要内容就是讲解消息的传递方式，上一节已经讲解完客户端和broker端连接的建立方式，在Connection、Session、Producer类对象建立的同时，客户端和broker端会进行一些消息交互，ActiveMQ中把所有的消息交互的内容都叫做Command，每条消息对应一个Command，例如客户端刚连接到broker，broker会发送一个BrokerInfo信息到客户端，接着客户端会发送ConnectionInfo连接信息、ProducerInfo生产者信息等等到服务端，如下图所示：
这些命令都继承于BaseCommand抽象类并实现于Command接口，类图如下，这里使用了访问者设计模式以及适配器设计模式。
适配器模式分为三种实现，第一种是通过继承实现，第二种是通过组合实现，第三种也就是类图中画的叫做默认适配器，CommandVisiter接口中定义了处理所有消息的方法，是broker和客户端api公用的一个接口，但是客户端用不到全部接口，如果实现这个接口那又必须实现全部的方法，所以此处在接口和具体类中间新增了一个CommandVisiterAdapter抽象类实现了全部的处理消息的方法并且全部都是空实现，这样在新建DefaultVisiter的时候就可以根据自己的需要来选择相应的方法进行实现了，访问者设计模式的体现请看下面的代码：
/**
* reads packets from a Socket
*/
publicvoid run() {
LOG.trace("TCP consumer thread for " + this + " starting");
this.runnerThread=Thread.currentThread();
try {
while (!isStopped()) {
doRun();
}
} catch (IOException e) {
stoppedLatch.get().countDown();
onException(e);
} catch (Throwable e){
stoppedLatch.get().countDown();
IOException ioe=new IOException("Unexpected error occured: " + e);
ioe.initCause(e);
onException(ioe);
}finally {
stoppedLatch.get().countDown();
}
}
protectedvoid doRun() throws IOException {
try {
Object command = readCommand();
doConsume(command);
} catch (SocketTimeoutException e) {
} catch (InterruptedIOException e) {
}
}
protected Object readCommand() throws IOException {
return wireFormat.unmarshal(dataIn);
}
这段代码是来自TcpTransport.java中，上一节已经讲解完TcpTransport的建立所以此处不在熬述，可以看到线程启动后一直在调用doRun方法，doRun方法则调用readCommand来读取客户端发送过来的信息，读到之后就会调用ActiveMQConnection.java的onCommand方法，如下代码：
@Override
publicvoid onCommand(final Object o) {
final Command command = (Command)o;
if (!closed.get() && command != null) {
try {
command.visit(new CommandVisitorAdapter() {
@Override
public Response processMessageDispatch(MessageDispatch md) throws Exception {
waitForTransportInterruptionProcessingToComplete();
ActiveMQDispatcher dispatcher = dispatchers.get(md.getConsumerId());
if (dispatcher != null) {
// Copy in case a embedded broker is dispatching via
// vm://
// md.getMessage() == null to signal end of queue
// browse.
Message msg = md.getMessage();
if (msg != null) {
msg = msg.copy();
msg.setReadOnlyBody(true);
msg.setReadOnlyProperties(true);
msg.setRedeliveryCounter(md.getRedeliveryCounter());
msg.setConnection(ActiveMQConnection.this);
msg.setMemoryUsage(null);
md.setMessage(msg);
}
dispatcher.dispatch(md);
}
returnnull;
}
@Override
public Response processProducerAck(ProducerAck pa) throws Exception {
if (pa != null && pa.getProducerId() != null) {
ActiveMQMessageProducer producer = producers.get(pa.getProducerId());
if (producer != null) {
producer.onProducerAck(pa);
}
}
returnnull;
}
@Override
public Response processBrokerInfo(BrokerInfo info) throws Exception {
brokerInfo = info;
brokerInfoReceived.countDown();
optimizeAcknowledge &= !brokerInfo.isFaultTolerantConfiguration();
getBlobTransferPolicy().setBrokerUploadUrl(info.getBrokerUploadUrl());
returnnull;
}
@Override
public Response processConnectionError(final ConnectionError error) throws Exception {
executor.execute(new Runnable() {
@Override
publicvoid run() {
onAsyncException(error.getException());
}
});
returnnull;
}
@Override
public Response processControlCommand(ControlCommand command) throws Exception {
onControlCommand(command);
returnnull;
}
@Override
public Response processConnectionControl(ConnectionControl control) throws Exception {
onConnectionControl((ConnectionControl)command);
returnnull;
}
@Override
public Response processConsumerControl(ConsumerControl control) throws Exception {
onConsumerControl((ConsumerControl)command);
returnnull;
}
@Override
public Response processWireFormat(WireFormatInfo info) throws Exception {
onWireFormatInfo((WireFormatInfo)command);
returnnull;
}
});
} catch (Exception e) {
onClientInternalException(e);
}
}
for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) {
TransportListener listener = iter.next();
listener.onCommand(command);
}
}
可以看到Command作为入参传进onCommand方法，然后方法中调用command.visit(new CommandVisitorAdapter() {…}),而new CommandVisitorAdapter就是上面介绍过的适配器设计模式，这里可以看到这个匿名类只需实现可以用到的方法，而不是实现所有的接口方法。这里command.visit的好处是传入进来的command程序不需要判断他是什么类型的command然后在决定调用这个command的某个方法，而是直接调用visit方法即可，而所有的业务逻辑也是统一的在CommandVisitorAdaptor中实现，这也算是java中动态多分派的实现。
知道了消息接收处理以及broker和客户端的信息交互之后，我们在来看下消息是如何从Command类序列化到字节写入以及字节如何反序列化转成Command类的，在ActiveMQ中，每一个Command消息都对应一个CommandMarshal类，例如ConnectionInfo使用ConnectionInfoMarshal来序列化和反序列化，BrokerInfo使用BrokerInfoMarshal来序列化和反序列化，ConnectionInfoMarshal等marshal的实现就是把字符串或者字节等信息写入到socket的输出流中没有什么可说的，序列化和反序列化又分为两种，一种是tight一种是loose，tight会针对cpu来进行优化，先写入大小，在写入具体数据，而loose方式则直接写入数据，两种方式都会使用缓存功能，客户端和服务端都分别存在marshal[]和unMarshal[]数组，例如客户端给服务端发送ProducerInfo信息，第一次发送后会把ProducerInfo存放在marshal的第0位，然后服务端接收后会把这个producerInfo放在unMarshal[0]中，如果客户端在次发送ProducerInfo则从缓存中取，找到ProducerInfo在marshal[]的第0个，则直接发送0，服务端则从unMarshal[]第0个取出使用，这节就到这里吧，想到什么我在补充，欢迎跟帖讨论。
这篇关于ActiveMQ源码架构解析第二节的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
