ActiveMQ producer同步/异步发送消息

http://activemq.apache.org/async-sends.html

producer发送消息有同步和异步两种模式，可以通过代码配置：

((ActiveMQConnection)connection).setUseAsyncSend(true);

producer默认是异步发送消息。在没有开启事务的情况下，producer发送持久化消息是同步的，调用send会阻塞直到broker把消息保存到磁盘并返回确认。

消息设置为持久：

MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);

消息设置为非持久：

MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

producer发送消息的调用栈如下：

// ActiveMQSession
protected void send(ActiveMQMessageProducer producer, ActiveMQDestination destination, Message message, int deliveryMode, int priority, long timeToLive,MemoryUsage producerWindow, int sendTimeout, AsyncCallback onComplete) throws JMSException {// 省略其他代码// 消息的持久类型和和连接模式是或的：所以只要connection配置为异步，就走异步发送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.increaseUsage(size);}} else { // 同步发送if (sendTimeout > 0 && onComplete==null) {this.connection.syncSendPacket(msg,sendTimeout);}else {this.connection.syncSendPacket(msg, onComplete);}}
}

 

producer发送同步消息的调用栈：

// org.apache.activemq.transport.ResponseCorrelator
public Object request(Object command) throws IOException {FutureResponse response = asyncRequest(command, null);return response.getResult();
}public FutureResponse asyncRequest(Object o, ResponseCallback responseCallback) throws IOException {Command command = (Command) o;command.setCommandId(sequenceGenerator.getNextSequenceId());// 需要回复command.setResponseRequired(true);FutureResponse future = new FutureResponse(responseCallback);IOException priorError = null;synchronized (requestMap) {priorError = this.error;if (priorError == null) {requestMap.put(new Integer(command.getCommandId()), future);}}if (priorError != null) {future.set(new ExceptionResponse(priorError));throw priorError;}next.oneway(command);return future;
}

 

producer发送异步消息的调用栈：

//org.apache.activemq.transport.ResponseCorrelator
public void oneway(Object o) throws IOException {Command command = (Command)o;command.setCommandId(sequenceGenerator.getNextSequenceId());// 不需要回复command.setResponseRequired(false);next.oneway(command);
}

 

在不考虑事务的情况下：

producer发送持久化消息是同步发送，发送是阻塞的，直到收到确认。同步发送肯定是有流量控制的。

producer默认是异步发送，异步发送不会等待broker的确认， 所以就需要考虑流量控制了：

ActiveMQConnectionFactory.setProducerWindowSize(int producerWindowSize)

ProducerWindowSize的含义：producer每发送一个消息，统计一下发送的字节数，当字节数达到ProducerWindowSize值时，需要等待broker的确认，才能继续发送。