| 编辑推荐: |
本文来自于csdn，本文重点分析使用JMS向activeMQ中间件发送消息的过程分析，希望对您的学习有所帮助。
|
|
同步发送与异步发送
activeMQ发送消息客户端发送消息分为同步发送与异步发送
同步发送，发送者发送一条消息会阻塞直到broker反馈一个确认消息给发送者，表示消息已经被broker处理。这个机
制提供了消息的安全性保障，由于是阻塞的操作，会影响到客户端消息发送的性能。
异步发送，发送者不需要等待broker提供反馈给发送者，性能相对较高。但是可能会出现消息丢失的情况。所
以使用异步发送的前提是在某些情况下允许出现数据丢失的情况。
那么在什么情况下选择同步发生，什么情况下是选择异步发送呢？
除去我们人为设置的方式，其默认的选择策略总结如下：
1.非持久化的消息都是异步发送的。
2.持久化消息在非事务模式下是同步发送的。
3.在开启事务的情况下，消息都是异步发送。
由于异步发送的效率会比同步发送性能更高，在发送持久化消息的时候，尽量去开启事务会话。或者设置使用异步发送。
以上三种默认策略，在源码中都会体现出来，稍后在源码中分析。
除了默认设置，我们可以认为设置发送模式。总结如下三种方式。
| 1.ConnectionFactory
connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.0.15:61616?
jms.useAsyncSend=true"); |
| 2.((ActiveMQConnectionFactory)
connectionFactory).setUseAsyncSend(true); |
| 3.((ActiveMQConnection)connection).setUseAsyncSend(true);
|
异步发送分析
首先分析异步发送过程，其数据流转如下图所示。
producerWindow
producerWindow主要是针对异步发送消息时，控制允许能够积压的消息累计大小，这些消息没有得到broker的ack，在得到ack之后会减去相应的消息size，释放producerWindow。
这个producerWindow的大小设置有2种方式
1.在brokerUrl中设置: “tcp : //localhost:61616?jms
. producerWindowSize = 1048576”,这种设置将会对所有的 producer生效。
2.在destinationUri中设置: “ test-queue ?producer.windowSize
= 1048576 ”,此参数只会对使用此 Destination 实例 的producer失效，将会覆盖brokerUrl中的producerWindowSize值。
通过设置producerWindow大小可以控制消息发送的流量控制。
下面开始跟踪源码分析图中的流程。
ActiveMQMessageProducer.send
发送方法跟踪，首先的关键方法就是这个，源码如下。
public void send(Destination
destination, Message message, int deliveryMode,
int priority, long timeToLive, AsyncCallback onComplete)
throws JMSException {
checkClosed();
if (destination == null) {
if (info.getDestination() == null) {
throw new UnsupportedOperationException("A
destination must be specified.");
}
throw new InvalidDestinationException("Don't
understand null destinations");
}
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.");
}
}
this.session.send(this, dest, message, deliveryMode,
priority, timeToLive, producerWindow, sendTimeout,
onComplete);
stats.onMessage();
}
|
上面的代码根据异步的流程图分析，看到是先需要根据producerWindow，来判断是否阻塞的，如果
producerWindow 不够，那么 producerWindow.waitForSpace();
就会阻塞等待。
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);
}
//获取会话的监视器
synchronized (sendMutex) {
// tell the Broker we are about to start a new
transaction
//开启事务
doStartTransaction();
TransactionId txid = transactionContext.getTransactionId();
long sequenceNumber = producer.getMessageSequence();
//......组装消息内容
//。。。。。。
//判断是一部还是同步发送
if (onComplete==null && sendTimeout <=
0 && !msg.isResponseRequired() &&
!connection.isAlwaysSyncSend() && (!msg.isPersistent()
|| connection.isUseAsyncSend() || txid != null))
{
this.connection.asyncSendPacket(msg);
if (producerWindow != null) {
// Since we defer lots of the marshaling till
we hit the
// wire, this might not
// provide and accurate size. We may change over
to doing
// more aggressive marshaling,
// to get more accurate sizes.. this is more important
once
// users start using producer window
// flow control.
int size = msg.getSize();
producerWindow.increaseUsage(size);
}
} else {
if (sendTimeout > 0 && onComplete==null)
{
this.connection.syncSendPacket(msg,sendTimeout);
}else {
this.connection.syncSendPacket(msg, onComplete);
}
}
}
}
|
上面这段代码，根据异步流程图来分析，主要看的就是判断是异步还是同步发送的代码，异步发送需要满足：
onComplete==null
&& sendTimeout <= 0 && !msg.isResponseRequired()
&& !connection.isAlwaysSyncSend() &&
(!msg.isPersistent() || connection.isUseAsyncSend()
|| txid != null
|
onComplete==null:发送回调为空。默认满足
sendTimeout <= 0：发送超时时间小于等于0。默认满足
!msg.isResponseRequired()：消息不需要响应。默认满足
!connection.isAlwaysSyncSend()：连接不能设置为同步发送。默认满足
(!msg.isPersistent() || connection.isUseAsyncSend()
|| txid != null)：是非持久化消息或者设置为使用异步发送或者开启了事务，从这里可以看出默认策略中的：
1.非持久化的消息都是异步发送的。!msg.isPersistent()
3.在开启事务的情况下，消息都是异步发送。txid != null
ActiveMQConnection.asyncSendPacket->doAsyncSendPacket
跟踪发送，到如下代码。
private void
doAsyncSendPacket(Command command) throws JMSException
{
try {
this.transport.oneway(command);
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}
|
上面代码的transport对象在定义里面没有进行构造，那么猜测应该是在创建ActiveMQConnection时，构造方法里面注入的，看一下注入的实例是什么。
ActiveMQConnectionFactory.createTransport
从connectionFactory.createConnection()方法一路跟踪，最后到了如下代码
protected Transport
createTransport() throws JMSException {
try {
URI connectBrokerUL = brokerURL;
//获取url中定义的连接模式
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);
}
}
|
来到上面这段代码，是根据我们brokerURL中的scheme来构建一个uri对象，然后用这个uri对象调用TransportFactory
.connect(connectBrokerUL) ;获取对应的transport实例。咋一看之下，有点像dubbo源码中利用url参数驱动，适配器根据url参数的不同，加载不同的实现类实例。接着跟踪TransportFactory
.connect(connectBrokerUL) ;看是不是这样。
public static
Transport connect(URI location) throws Exception
{
TransportFactory tf = findTransportFactory(location);
return tf.doConnect(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;
}
|
可以看出是在(TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme);创建实例的，查看TRANSPORT_FACTORY_FINDER的定义如下。
| private static
final FactoryFinder TRANSPORT_FACTORY_FINDER =
new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
|
在newInstance方法中，代码如下
public Object
newInstance(String key) throws IllegalAccessException,
InstantiationException, IOException, ClassNotFoundException
{
return objectFactory.create(path+key);
}
|
上面的path就是”“ META-INF/services/org/apache/activemq/transport/”“
, key在我们的brokerUrl就是tcp，猜测
就是找一个在该路径下的文件中定义的类全路径作为实例化的类型信息，查看路径下的文件信息如
还真有一个tcp文件，打开看
class=org.apache.activemq.transport.tcp.
TcpTransportFactory |
后续能估计到就是实例化一个TcpTransportFactory，然后回到下面的代码。
public static
Transport connect(URI location) throws Exception
{
TransportFactory tf = findTransportFactory(location);
return tf.doConnect(location);
}
|
TcpTransportFactory.doConnect
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);
//包装Transport
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);
}
}
|
上面代码就执行了真正的创建Transport，同时给Transport包装了外城处理逻辑，具体包装在，如下代码
public Transport
configure(Transport transport,
WireFormat wf,
Map options) throws Exception {
transport = compositeConfigure(transport, wf,
options);
transport = new MutexTransport(transport);
transport = new ResponseCorrelator(transport);
return transport;
}
public Transport compositeConfigure(Transport
transport,
WireFormat format, Map options) {
//如果配置了写超时则执行包装WriteTimeoutFilter
if (options.containsKey(WRITE_TIMEOUT_FILTER))
{
transport = new WriteTimeoutFilter(transport);
String soWriteTimeout = (String)options.remove
(WRITE_TIMEOUT_FILTER);
if (soWriteTimeout!=null) {
((WriteTimeoutFilter)transport).setWriteTimeout
(Long.parseLong
(soWriteTimeout));
}
}
IntrospectionSupport.setProperties(transport,
options);
return transport;
}
|
所以随后执行oneway的Transport实例是这样的：ResponseCorrelator（MutexTransport（TcpTransport）），在调用时就会构成调用链条逐层调用。
ResponseCorrelator.oneway
| public void
oneway(Object o) throws IOException {
Command command = (Command)o;
command.setCommandId(sequenceGenerator.
getNextSequenceId());
command.setResponseRequired(false);
next.oneway(command);
} |
MutexTransport.oneway
负责枷锁
@Override
public void oneway(Object command) throws IOException
{
writeLock.lock();
try {
next.oneway(command);
} finally {
writeLock.unlock();
}
}
|
TcpTransport.oneWay
异步发送消息
@Override
public void oneway(Object command) throws IOException
{
checkStarted();
wireFormat.marshal(command, dataOut);
dataOut.flush();
}
|
这种链式代码风格值得学习，逻辑抽出也很巧妙的分布在不同的处理器中。
差点忘了，最后还要增加producerWindowSize
if (producerWindow
!= null) {
// Since we defer lots of the marshaling till
we hit the
// wire, this might not
// provide and accurate size. We may change over
to doing
// more aggressive marshaling,
// to get more accurate sizes.. this is more important
once
// users start using producer window
// flow control.
int size = msg.getSize();
producerWindow.increaseUsage(size);
}
|
然后，异步发送的流程搞定。
同步发送消息分析
ActiveMqConnection.syncSendPacket
public Response
syncSendPacket(Command command, int timeout) throws
JMSException {
if (isClosed()) {
throw new ConnectionClosedException();
} else {
try {
//调用Transport的request'方法
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();
} else {
if (isClosed()||closing.get()) {
LOG.debug("Received an exception but connection
is closing");
}
JMSException jmsEx = null;
try {
jmsEx = JMSExceptionSupport.create(er.getException());
} catch(Throwable e) {
LOG.error("Caught an exception trying to
create a JMSException for " +er.getException(),e);
}
if (er.getException() instanceof SecurityException
&& command instanceof ConnectionInfo){
forceCloseOnSecurityException(er.getException());
}
if (jmsEx !=null) {
throw jmsEx;
}
}
}
return response;
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}
}
|
在异步分析时已经知道Transport是一个调用链，首先来到的是
ResponseCorrelator.request
public Object
request(Object command) throws IOException {
FutureResponse response = asyncRequest(command,
null);
return response.getResult();
}
|
然后你会发现，其实同步发送也是先执行异步发送请求asyncRequest，然后再阻塞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, this);
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;
}
|
|
