ActiveMQ消息的发送原理
持久化消息和非持久化消息的发送策略：消息同步发送和异步发送
ActiveMQ支持同步、异步两种发送模式将消息发送到broker上。同步发送过程当中，发送者发送一条消息会阻塞直到broker反馈一个确认消息，表示消息已经被broker处理。这个机制提供了消息的安全性保障，可是因为是阻塞的操做，会影响到客户端消息发送的性能。异步发送的过程当中，发送者不须要等待broker提供反馈，因此性能相对较高。可是可能会出现消息丢失的状况。因此使用异步发送的前提是在某些状况下容许出现数据丢失的状况。java
默认状况下，非持久化消息是异步发送的，持久化消息而且是在非事务模式下是同步发送的。可是在开启事务的状况下，消息都是异步发送。因为异步发送的效率会比同步发送性能更高。因此在发送持久化消息的时候，尽可能去开启事务会话。除了持久化消息和非持久化消息的同步和异步特性之外，咱们还能够经过如下几种方式来设置异步发送：apache
1.ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.11.153:61616?jms.useAsyncSend=true");
2.((ActiveMQConnectionFactory) connectionFactory).setUseAsyncSend(true);
3.((ActiveMQConnection)connection).setUseAsyncSend(true);
消息发送的源码：
以producer.send为入口 进入的是ActiveMQSession 实现：缓存
public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, AsyncCallback onComplete) throws JMSException {
checkClosed(); //检查session的状态，若是session关闭则抛异常
if (destination == null) {
if (info.getDestination() == null) {
throw new UnsupportedOperationException("A destination must be specified.");
}
throw new InvalidDestinationException("Don't understand null destinations");
}
//检查destination的类型，若是符合要求，就转变为ActiveMQDestination
ActiveMQDestination dest;
if (destination.equals(info.getDestination())) {
dest = (ActiveMQDestination)destination;
} else if (info.getDestination() == null) {
dest = ActiveMQDestination.transform(destination);
} else {
throw new UnsupportedOperationException("This producer can only send messages to: " + this.info.getDestination().getPhysicalName());
}
if (dest == null) {
throw new JMSException("No destination specified");
}
if (transformer != null) {
Message transformedMessage = transformer.producerTransform(session, this, message);
if (transformedMessage != null) {
message = transformedMessage;
}
}
//若是发送窗口大小不为空，则判断发送窗口的大小决定是否阻塞
if (producerWindow != null) {
try {
producerWindow.waitForSpace();
} catch (InterruptedException e) {
throw new JMSException("Send aborted due to thread interrupt.");
}
}
//发送消息到broker的topic
this.session.send(this, dest, message, deliveryMode, priority, timeToLive, producerWindow, sendTimeout, onComplete);
stats.onMessage();
}
ActiveMQSession的send方法，this.session.send(this, dest, message, deliveryMode, priority, timeToLive, producerWindow, sendTimeout, onComplete)：安全
protected void send(ActiveMQMessageProducer producer, ActiveMQDestination destination, Message message, int deliveryMode, int priority, long timeToLive,
MemoryUsage producerWindow, int sendTimeout, AsyncCallback onComplete) throws JMSException {
checkClosed();
if (destination.isTemporary() && connection.isDeleted(destination)) {
throw new InvalidDestinationException("Cannot publish to a deleted Destination: " + destination);
}
//互斥锁，若是一个session的多个producer发送消息到这里，会保证消息发送的有序性
synchronized (sendMutex) {
// tell the Broker we are about to start a new transaction
doStartTransaction();//告诉broker开始一个新事务，只有事务型会话中才会开启
TransactionId txid = transactionContext.getTransactionId();//从事务上下文中获取事务id
long sequenceNumber = producer.getMessageSequence();
//Set the "JMS" header fields on the original message, see 1.1 spec section 3.4.11
message.setJMSDeliveryMode(deliveryMode); //在JMS协议头中设置是否持久化标识
long expiration = 0L;//计算消息过时时间
if (!producer.getDisableMessageTimestamp()) {
long timeStamp = System.currentTimeMillis();
message.setJMSTimestamp(timeStamp);
if (timeToLive > 0) {
expiration = timeToLive + timeStamp;
}
}
message.setJMSExpiration(expiration);//设置消息过时时间
message.setJMSPriority(priority);//设置消息的优先级
message.setJMSRedelivered(false);;//设置消息为非重发
// transform to our own message format here
ActiveMQMessage msg = ActiveMQMessageTransformation.transformMessage(message, connection);
msg.setDestination(destination);
msg.setMessageId(new MessageId(producer.getProducerInfo().getProducerId(), sequenceNumber));
// Set the message id.
if (msg != message) {//若是消息是通过转化的，则更新原来的消息id和目的地
message.setJMSMessageID(msg.getMessageId().toString());
// Make sure the JMS destination is set on the foreign messages too.
message.setJMSDestination(destination);
}
//clear the brokerPath in case we are re-sending this message
msg.setBrokerPath(null);
msg.setTransactionId(txid);
if (connection.isCopyMessageOnSend()) {
msg = (ActiveMQMessage)msg.copy();
}
msg.setConnection(connection);
msg.onSend();//把消息属性和消息体都设置为只读，防止被修改
msg.setProducerId(msg.getMessageId().getProducerId());
if (LOG.isTraceEnabled()) {
LOG.trace(getSessionId() + " sending message: " + msg);
}
//若是onComplete没有设置(这里传进来就是null)，且发送超时时间小于0，且消息不须要反馈，且链接器不是同步发送模式，且消息非持久化或者链接器是异步发送模式
//或者存在事务id的状况下，走异步发送，不然走同步发送
if (onComplete==null && sendTimeout <= 0 && !msg.isResponseRequired() && !connection.isAlwaysSyncSend() && (!msg.isPersistent() || connection.isUseAsyncSend() || txid != null)) {
this.connection.asyncSendPacket(msg);
if (producerWindow != null) {
int size = msg.getSize();//异步发送的状况下，须要设置producerWindow的大小
producerWindow.increaseUsage(size);
}
} else {
if (sendTimeout > 0 && onComplete==null) {
this.connection.syncSendPacket(msg,sendTimeout);//带超时时间的同步发送
}else {
this.connection.syncSendPacket(msg, onComplete);//带回调的同步发送
}
}
}
}
咱们从上面的代码能够看到，在执行发送操做以前须要把消息作一个转化，而且将咱们设置的一些属性注入导指定的属性中，咱们先来看看异步发送，会发现异步发送的时候涉及到producerWindowSize的大小：session
ProducerWindowSize的含义异步
producer每发送一个消息，统计一下发送的字节数，当字节数达到ProducerWindowSize值时，须要等待broker的确认，才能继续发送。async
主要用来约束在异步发送时producer端容许积压的(还没有ACK)的消息的大小，且只对异步发送有意义。每次发送消息以后，都将会致使memoryUsage大小增长(+message.size)，当broker返回producerAck时，memoryUsage尺寸减小(producerAck.size，此size表示先前发送消息的大小)。tcp
能够经过以下2种方式设置:
Ø 在brokerUrl中设置: "tcp://localhost:61616?jms.producerWindowSize=1048576",这种设置将会对全部的producer生效。
Ø 在destinationUri中设置: "myQueue?producer.windowSize=1048576",此参数只会对使用此Destination实例的producer生效，将会覆盖brokerUrl中的producerWindowSize值。
注意：此值越大，意味着消耗Client端的内存就越大。性能
接下去咱们进入异步发送流程，看看消息是怎么异步发送的this.connection.asyncSendPacket(msg)：ui
private void doAsyncSendPacket(Command command) throws JMSException {
try {
this.transport.oneway(command);
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}
这里的 Command 其实就是以前一步所转化的message ，而且通过一系列的属性注入。由于ActiveMQMessage 继承了 baseCommand ，该类实现了 Command 。因此能够转化，而后咱们发现 oneway 方法又不少的实现，都是基于 transport ，那么咱们就须要来看看这个 transport 是什么。这里咱们把代码往前翻并无发现他的初始化，按照咱们以往的思路，这里就会在初始化链接的时候进行初始化该对象：
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.254.135:61616");
Connection connection= connectionFactory.createConnection();
这里进入 ActiveMQConnectionFactory 的 createConnection方法会来到：
protected ActiveMQConnection createActiveMQConnection(String userName, String password) throws JMSException {
if (brokerURL == null) {
throw new ConfigurationException("brokerURL not set.");
}
ActiveMQConnection connection = null;
try {// 果真发现了这个东东的初始化
Transport transport = createTransport();
// 建立链接
connection = createActiveMQConnection(transport, factoryStats);
// 设置用户密码
connection.setUserName(userName);
connection.setPassword(password);
// 对链接作包装
configureConnection(connection);
// 启动一个后台传输线程
transport.start();
// 设置客户端消费的id
if (clientID != null) {
connection.setDefaultClientID(clientID);
}
return connection;
} ......
}
这里咱们发现了 Transport transport = createTransport(); 这就是他的初始化：咱们能够发现
protected Transport createTransport() throws JMSException {
try {
URI connectBrokerUL = brokerURL;
String scheme = brokerURL.getScheme();
if (scheme == null) {
throw new IOException("Transport not scheme specified: [" + brokerURL + "]");
}
if (scheme.equals("auto")) {
connectBrokerUL = new URI(brokerURL.toString().replace("auto", "tcp"));
} else if (scheme.equals("auto+ssl")) {
connectBrokerUL = new URI(brokerURL.toString().replace("auto+ssl", "ssl"));
} else if (scheme.equals("auto+nio")) {
connectBrokerUL = new URI(brokerURL.toString().replace("auto+nio", "nio"));
} else if (scheme.equals("auto+nio+ssl")) {
connectBrokerUL = new URI(brokerURL.toString().replace("auto+nio+ssl", "nio+ssl"));
}
return TransportFactory.connect(connectBrokerUL);
} catch (Exception e) {
throw JMSExceptionSupport.create("Could not create Transport. Reason: " + e, e);
}
}
这里有点相似于基于URL驱动的意思，这里进来先是构建一个 URI ，根据URL去建立一个链接TransportFactory.connect，会发现默认使用的是tcp的协议。这里因为咱们在建立链接的时候就已经指定了tcp因此这里的判断都没用，直接进入建立链接TransportFactory.connect(connectBrokerUL)：
public static Transport connect(URI location) throws Exception {
TransportFactory tf = findTransportFactory(location);
return tf.doConnect(location);
}
这里作链接须要建立一个 tf 对象。这就要看看findTransportFactory(location) ：
public static TransportFactory findTransportFactory(URI location) throws IOException {
String scheme = location.getScheme();
if (scheme == null) {
throw new IOException("Transport not scheme specified: [" + location + "]");
}
TransportFactory tf = TRANSPORT_FACTORYS.get(scheme);
if (tf == null) {
// Try to load if from a META-INF property.
try {
tf = (TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme);
TRANSPORT_FACTORYS.put(scheme, tf);
} catch (Throwable e) {
throw IOExceptionSupport.create("Transport scheme NOT recognized: [" + scheme + "]", e);
}
}
return tf;
}
不难理解以上的 代码是根据 scheme经过TRANSPORT_FACTORYS 这个map 来建立的 TransportFactory ，若是获取不到，就会经过TRANSPORT_FACTORY_FINDER 去获取一个实例。TRANSPORT_FACTORY_FINDER 这个FINDER是什么东西呢？ 咱们看看他的初始化：
private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
咱们经过源码中指定路径如下的东西：
这有点相似于 java 中SPI规范的意思。咱们能够看看 tcp 其中的内容：
class=org.apache.activemq.transport.tcp.TcpTransportFactory
这里是键值对的方式，上述获取实例的代码中其实就是获取一个 TcpTransportFactory 实例，那么咱们就知道tf.doConnect(location) 是哪一个实现类作的，就是TcpTransportFactory，可是咱们点开一看并未发现 TcpTransportFactory实现，这就说明该类使用的是父类里面的方法，这里就是TransportFactory 类：
public Transport doConnect(URI location) throws Exception {
try {
Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location));
if( !options.containsKey("wireFormat.host") ) {
options.put("wireFormat.host", location.getHost());
}
WireFormat wf = createWireFormat(options);
//建立一个Transport 这里才是咱们要找的真相
Transport transport = createTransport(location, wf);
//配置configure，这个里面是对Transport作链路包装，思想相似于dubbo的cluster
Transport rc = configure(transport, wf, options);
//remove auto
IntrospectionSupport.extractProperties(options, "auto.");
if (!options.isEmpty()) {
throw new IllegalArgumentException("Invalid connect parameters: " + options);
}
return rc;
} catch (URISyntaxException e) {
throw IOExceptionSupport.create(e);
}
}
咱们进入 createTransport(location, wf) 方法，这里是使用Tcp子类的实现。会发现里面建立了一个 Sokect 链接 ，这就是准备后来进行发送的Sokect。而后这里返回的 Transport 就是 TcpTransport .接下去就是对这个 transport 进行包装 configure(transport, wf, options)：
public Transport configure(Transport transport, WireFormat wf, Map options) throws Exception {
//组装一个复合的transport，这里会包装两层，一个是IactivityMonitor.另外一个是WireFormatNegotiator
transport = compositeConfigure(transport, wf, options);
//再作一层包装,MutexTransport
transport = new MutexTransport(transport);
//包装ResponseCorrelator
transport = new ResponseCorrelator(transport);
return transport;
}
到目前为止，这个transport实际上就是一个调用链了，他的链结构为ResponseCorrelator(MutexTransport(WireFormatNegotiator(IactivityMonitor(TcpTransport()))每一层包装表示什么意思呢？
ResponseCorrelator 用于实现异步请求。
MutexTransport 实现写锁，表示同一时间只容许发送一个请求
WireFormatNegotiator 实现了客户端链接broker的时候先发送数据解析相关的协议信息，好比解析版本号，是否使用缓存等
InactivityMonitor 用于实现链接成功成功后的心跳检查机制，客户端每10s发送一次心跳信息。服务端每30s读取一次心跳信息。
经过这层层的分析，咱们回到 ActiveMQConnection 发送消息的doAsyncSendPacket 方法：
private void doAsyncSendPacket(Command command) throws JMSException {
try {
this.transport.oneway(command);
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}
这里的 oneway(command)方法会前后经历上述调用链的处理最后调用到 TcpTransport 的oneway(command) ，咱们一步一步来看看都作了些什么：
ResponseCorrelator.oneway(command):里面就设置了两个属性
public void oneway(Object o) throws IOException {
Command command = (Command)o; //对前面的对象作一个强转，组装一些信息
command.setCommandId(sequenceGenerator.getNextSequenceId());
command.setResponseRequired(false);
next.oneway(command);
}
MutexTransport.oneway(command):
public void oneway(Object command) throws IOException {
writeLock.lock();// 经过 ReentrantLock作加锁
try {
next.oneway(command);
} finally {
writeLock.unlock();
}
}
WireFormatNegotiator.oneway(command):这个里面调用了父类的 oneway ，父类是 TransportFilter 类
public void oneway(Object command) throws IOException {
boolean wasInterrupted = Thread.interrupted();
try {
if (readyCountDownLatch.getCount() > 0 && !readyCountDownLatch.await(negotiateTimeout, TimeUnit.MILLISECONDS)) {
throw new IOException("Wire format negotiation timeout: peer did not send his wire format.");
}
} catch (InterruptedException e) {
InterruptedIOException interruptedIOException = new InterruptedIOException("Interrupted waiting for wire format negotiation");
interruptedIOException.initCause(e);
try {
onException(interruptedIOException);
} finally {
Thread.currentThread().interrupt();
wasInterrupted = false;
}
throw interruptedIOException;
} finally {
if (wasInterrupted) {
Thread.currentThread().interrupt();
}
}
super.oneway(command); //里面没作什么事情进入下一个调用链
}
从WireFormatNegotiator的父类TransportFilter进入下一个调用链应该调用的是InactivityMonitor.oneway(command)，但是并未发现又该类实现，因此这里进入InactivityMonitor 的父类AbstractInactivityMonitor：
public void oneway(Object o) throws IOException {
// To prevent the inactivity monitor from sending a message while we
// are performing a send we take a read lock. The inactivity monitor
// sends its Heart-beat commands under a write lock. This means that
// the MutexTransport is still responsible for synchronizing sends
sendLock.readLock().lock();//获取发送读锁 锁定
inSend.set(true);//设置属性
try {
doOnewaySend(o);//经过这个逻辑进入下一个调用链
} finally {
commandSent.set(true);
inSend.set(false);
sendLock.readLock().unlock();
}
}
在doOnewaySend 里面的next.oneway(command) 方法最终调用 TcpTransport 的实现：
public void oneway(Object command) throws IOException {
checkStarted();
//进行格式化内容 经过Sokct 发送
wireFormat.marshal(command, dataOut);
// 流的刷新
dataOut.flush();
}
最后经过Sokect进行数据的传输。这样子异步发送的流程就结束了。下面来走一下同步的流程：经过this.connection.syncSendPacket() 进入同步发送流程。
public Response syncSendPacket(Command command, int timeout) throws JMSException {
if (isClosed()) {
throw new ConnectionClosedException();
} else {
try {// 进行发送，阻塞获取结果
Response response = (Response)(timeout > 0
? this.transport.request(command, timeout)
: this.transport.request(command));
if (response.isException()) {
ExceptionResponse er = (ExceptionResponse)response;
if (er.getException() instanceof JMSException) {
throw (JMSException)er.getException();
}
。。。。。。。。。
return response;
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}
}
这里的 transport 跟异步发送过程当中的transport时同样的，即 ResponseCorrelator(MutexTransport(WireFormatNegotiator(IactivityMonitor(TcpTransport())) 一个调用链，进入ResponseCorrelator 的实现：
public Object request(Object command, int timeout) throws IOException {
FutureResponse response = asyncRequest(command, null);
return response.getResult(timeout);
}
从这个方法咱们能够获得的信息时，在发送的时候采用的是 asyncRequest 方法，意思是异步请求，可是在下行采用 response.getResult(timeout) 去同步阻塞的方式去获取结果：
public Response getResult(int timeout) throws IOException {
final boolean wasInterrupted = Thread.interrupted();
try {
Response result = responseSlot.poll(timeout, TimeUnit.MILLISECONDS);
.........
}
这里会从 ArrayBlockingQueue 去 阻塞的处理请求。其实这里的同步发送实质上采用的不阻塞发送，阻塞的去等待broker 的反馈结果。
最后整理一下这个发送流程图
- 1. php 事务处理，ActiveMQ的发送消息，与处理消息 php ActiveMQ的发送消息，与处理消息
- 2. ActiveMQ发送、接收消息
- 3. activeMQ消息发送过程与原理浅析
- 4. 分布式消息通讯ActiveMQ原理-发送策略-笔记
- 5. ActiveMQ 部署及发送接收消息
- 6. Spring + ActiveMQ实现jms发送消息
- 7. ActiveMQ集成Spring并发送消息
- 8. activemq+spring 持久化发送消息
- 9. activemq结合mqtt发送p2p消息
- 10. iOS开发消息推送原理
- 更多相关文章...
- • HTTP 消息结构 - HTTP 教程
- • MyBatis的工作原理 - MyBatis教程
- • ☆技术问答集锦（13）Java Instrument原理
- • Java Agent入门实战（三）-JVM Attach原理与使用
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 融合阿里云，牛客助您找到心仪好工作
- 2. 解决jdbc(jdbctemplate)在测试类时不报错在TomCatb部署后报错
- 3. 解决PyCharm GoLand IntelliJ 等 JetBrains 系列 IDE无法输入中文
- 4. vue+ant design中关于图片请求不显示的问题。
- 5. insufficient memory && Native memory allocation (malloc) failed
- 6. 解决IDEA用Maven创建的Web工程不能创建Java Class文件的问题
- 7. [已解决] Error: Cannot download ‘https://start.spring.io/starter.zip?
- 8. 在idea让java文件夹正常使用
- 9. Eclipse启动提示“subversive connector discovery”
- 10. 帅某-技巧-快速转帖博主文章（article_content）