英文:
Connection.start is not needed for JMS MessageProducer but needed for MessageConsumer
问题
<h1>A - 问题</h1>
我知道在 Stack Overflow 上有一个类似的问题，但不完全相同。
我试图理解 JMS 中 **MessageProducer** 和 **MessageConsumer** 的底层工作原理。我使用 **ActiveMQ** 的实现，编写了一个简单的 **MessageProducer** 示例，用于向队列发送消息，以及一个 **MessageConsumer** 示例，用于从队列消费消息，同时在本地运行 **ActiveMQ**。
对于发送消息到队列，需要使用 **Connection#start** 方法。具体的调试点如下所示。调用 **Connection#start** 会触发 **ActiveMQSession#start** 方法。当调用 **Connection#start** 时，会触发此方法。请参见下面的调试点 `org.apache.activemq.ActiveMQSession#start`；
[![ActiveMQ 调试点][1]][1]
问题是，**MessageProducer** 上不需要显式使用 **Connection#start**，但是在 **MessageConsumer** 上需要。然而，对于这两个示例，我们都需要清除资源（**session** 和 **connection**）。我意识到，如果我在生产者上删除 **Connection#start** 方法，代码将会执行，调试点不会被触发（甚至在底层也不会触发），我可以在队列中看到消息。但是，如果我在消费者上删除 **Connection#start** 方法，代码将不会执行，这就是问题所在，为什么在 **MessageProducer** 上不需要它，代码可以成功执行，但是在 **MessageConsumer** 上却需要？另外，为什么即使我们不使用 **Connection#start** 来发送消息，即使我们需要关闭连接以刷新资源。这似乎有些不合理。
我看到 **started** 字段是一个 `AtomicBoolean` 类型。我不是并发和多线程方面的专家，所以可能有人可以解释一下为什么对于 **MessageProducer**，不需要强制执行 Connection#start；
[![org.apache.activemq.ActiveMQSession - started 字段][2]][2]
<h1>B - 使用 ActiveMQ 的 JMS MessageProducer 示例代码</h1>
```java
package com.bzdgn.jms.stackoverflow;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSSendMessageToQueue {
private static final String ACTIVE_MQ_URL = "tcp://localhost:61616";
public static void main(String[] args) throws JMSException {
String queueName = "test_queue";
String messageContent = "Hello StackOverflow!";
// ActiveMQ 实现的 Connection Factory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ACTIVE_MQ_URL);
// 从 Connection Factory 获取连接
Connection connection = connectionFactory.createConnection();
// 创建会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 发送消息到队列
Queue queue = session.createQueue(queueName);
TextMessage msg = session.createTextMessage(messageContent);
MessageProducer messageProducer = session.createProducer(queue);
messageProducer.send(msg);
// 清除资源
session.close();
connection.close();
}
}
<h1>C - 使用 ActiveMQ 的 JMS MessageConsumer 示例代码</h1>
package com.bzdgn.jms.stackoverflow;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSConsumeMessageFromQueue {
private static final String ACTIVE_MQ_URL = "tcp://localhost:61616";
public static void main(String[] args) throws JMSException {
String queueName = "test_queue";
// ActiveMQ 实现的 Connection Factory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ACTIVE_MQ_URL);
// 从 Connection Factory 获取连接
Connection connection = connectionFactory.createConnection();
// 创建会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 从队列消费消息
Queue queue = session.createQueue(queueName);
MessageConsumer messageConsumer = session.createConsumer(queue);
connection.start();
Message message = messageConsumer.receive(500);
if (message != null) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
String messageContent = textMessage.getText();
System.out.println("Message Content: " + messageContent);
}
} else {
System.out.println("No message in the queue: " + queueName);
}
// 清除资源
session.close();
connection.close();
}
}
<h1>D - 配置和 Maven 依赖</h1>
JDK 版本为 1.8，我正在运行 ActiveMQ 5.15.12，并且客户端依赖项也使用了相同的版本；
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.15.12</version>
</dependency>
<details>
<summary>英文:</summary>
<h1>A - Question</h1>
I know there is a similar question but not the same in SO.
I'm trying to understand what goes on under the hood with **MessageProducer** and **MessageConsumer** in JMS. Using the implementation of **ActiveMQ**, I've written a simple **MessageProducer** example to send a message to queue, and a **MessageConsumer** example to consume the message from the queue, while running **ActiveMQ** locally.
**Connection#start** method is needed for sending a Message to Queue. The exact debug point is as follows. **Connection#start** triggers **ActiveMQSession#start** method. This method is triggered when a **Connection#start** is called. See the following debug point at `org.apache.activemq.ActiveMQSession#start`;
[![ActiveMQ Debug Point][1]][1]
The problem is, **Connection#start** is not explicitly needed on a **MessageProducer** but needed on a **MessageConsumer**. However, for both examples, we need to clear the resources (**session** and **connection**). What I realized is, if I remove **Connection#start** method on producer, the code will execute, debug point won't be triggered (not even under the hood) and I see the message in the queue. But if I remove Connection#start method on consumer, the code won't execute, that's the question, why it's not needed in **MessageProducer** and the code executes successfully but needed on **MessageConsumer**? Also why even we don't use **Connection#start** for **MessageProducer** even to the fact that we need to close the connection in order to flush the resources. It seems like code smells.
I see that field **started** is an `AtomicBoolean`. I'm not an expert on concurrency and multi-threading, so, may be there is a logic someone can explain why for a MessageProducer, a Connection#start is not mandatory;
[![org.apache.activemq.ActiveMQSession - started field][2]][2]
<h1>B - Example Code for JMS MessageProducer with ActiveMQ</h1>
package com.bzdgn.jms.stackoverflow;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSSendMessageToQueue {
private static final String ACTIVE_MQ_URL = "tcp://localhost:61616";
public static void main(String[] args) throws JMSException {
String queueName = "test_queue";
String messageContent = "Hello StackOverflow!";
// Connection Factory from ActiveMQ Implementation
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ACTIVE_MQ_URL);
// Get connection from Connection Factory
Connection connection = connectionFactory.createConnection();
// Create session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Send Message to Queue
Queue queue = session.createQueue(queueName);
TextMessage msg = session.createTextMessage(messageContent);
MessageProducer messageProducer = session.createProducer(queue);
messageProducer.send(msg);
// Clear resources
session.close();
connection.close();
}
}
<h1>C - Example Code for JMS MessageConsumer with ActiveMQ</h1>
package com.bzdgn.jms.stackoverflow;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class JMSConsumeMessageFromQueue {
private static final String ACTIVE_MQ_URL = "tcp://localhost:61616";
public static void main(String[] args) throws JMSException {
String queueName = "test_queue";
// Connection Factory from ActiveMQ Implementation
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ACTIVE_MQ_URL);
// Get connection from Connection Factory
Connection connection = connectionFactory.createConnection();
// Create session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Consume Message from the Queue
Queue queue = session.createQueue(queueName);
MessageConsumer messageConsumer = session.createConsumer(queue);
connection.start();
Message message = messageConsumer.receive(500);
if ( message != null ) {
if ( message instanceof TextMessage ) {
TextMessage textMessage = (TextMessage) message;
String messageContent = textMessage.getText();
System.out.println("Message Content: " + messageContent);
}
} else {
System.out.println("No message in the queue: " + queueName);
}
// Clear resources
session.close();
connection.close();
}
}
<h1>D - Configuration And Maven Dependency</h1>
**JDK** version is `1.8`, I'm running `ActiveMQ 5.15.12` and also using the same version for the client dependency;
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.15.12</version>
</dependency>
[1]: https://i.stack.imgur.com/VWKVl.png
[2]: https://i.stack.imgur.com/tdtGh.png
</details>
# 答案1
**得分**: 2
这里的行为受JMS规范的规定。简而言之，`javax.jms.Connection.start()` 方法适用于消费者而不是生产者。它通知代理开始向与连接关联的消费者传递消息。关于 `Connection` 的[JavaDoc][1]中写道：
> 通常情况下，直到设置完成（也就是直到所有消息消费者都已创建），才将连接保持在停止模式。在这一点上，客户端调用连接的 start 方法，消息开始到达连接的消费者。这种设置约定可以在客户端设置自身时最大程度地减少因异步消息传递而导致的任何客户端混淆。
>
> 连接可以立即启动，然后进行设置。这样做的客户端必须准备好在仍在设置过程中处理异步消息传递。
`start()` 方法对生产者没有影响。你看到的是预期的行为。
值得注意的是，如果你使用的是JMS 2的简化API，则此行为稍有不同。如果您使用 `JMSContext` 来创建 `JMSConsumer`，则消息传递会自动启动。需要明确的是，ActiveMQ 5.x 不实现JMS 2，但是 [ActiveMQ Artemis][2] 实现了。
[1]: https://docs.oracle.com/javaee/7/api/javax/jms/Connection.html
[2]: http://activemq.apache.org/components/artemis/
<details>
<summary>英文:</summary>
The behavior here is dictated by the JMS specification. Simply put, `javax.jms.Connection.start()` applies to consumers not producers. It tells the broker to begin delivering messages to the consumers associated with the connection. The [JavaDoc for `Connection`][1] says this:
> It is typical to leave the connection in stopped mode until setup is complete (that is, until all message consumers have been created). At that point, the client calls the connection's start method, and messages begin arriving at the connection's consumers. This setup convention minimizes any client confusion that may result from asynchronous message delivery while the client is still in the process of setting itself up.
>
> A connection can be started immediately, and the setup can be done afterwards. Clients that do this must be prepared to handle asynchronous message delivery while they are still in the process of setting up.
The `start()` method has no impact on producers. You're seeing the expected behavior.
It's worth noting that this behavior is a bit different if you're using the simplified API which is part of JMS 2. If you use a `JMSContext` to create the a `JMSConsumer` then message delivery starts automatically. To be clear, ActiveMQ 5.x doesn't implement JMS 2, but [ActiveMQ Artemis][2] does.
[1]: https://docs.oracle.com/javaee/7/api/javax/jms/Connection.html
[2]: http://activemq.apache.org/components/artemis/
</details>
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论