ActiveMQ发送queue消息

痛定思痛。 2022-06-05 00:42 291阅读 0赞

我们在taotao-manager和taotao-search工程中都会用到消息服务，因此我们可以选择其中任何一个工程来进行测试，这里我们使用taotao-manager-service工程（taotao-manager的子工程）来测试一下面对面发送消息和接收消息。   
首先，我们需要在taotao-manager-service工程中添加对activemq的maven依赖，如下图所示，activemq的版本号在taotao-parent工程中已经统一定义好了，我们使用的是5.11.2版本的Jar包，之所以选择这个版本的Jar包，是因为这个版本的Jar包还没有spring相关的包结构，与spring结合不会有问题，我们安装的activemq服务是5.12.0版本的，它里面的Jar包是有spring相关的Jar包的，这样会引起Jar包冲突，出现各种各样的问题，因此我们虽然安装的是5.12.0版本的activemq，但是我们使用的Jar包却是5.11.2版本。   
![这里写图片描述][SouthEast]   
下面我们在taotao-manager-service工程中新建一个测试包和单元测试类，如下图所示。   
![这里写图片描述][SouthEast 1]   
为方便大家复制，现将TestActiceMQ单元测试类的代码贴出。

public class TestActiceMQ { 
    
        @Test
        public void testQueueProducer() throws Exception {
            /* * 1.创建一个连接工厂对象——ConnectionFactory对象，需要指定ActiceMQ服务的ip及端口号。 * 注意参数brokerURL的开头是tcp://，而不是我们通常所见的http://，端口是61616而不是我们访问ActiceMQ * 后台管理页面时所使用的8161。 */
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.129:61616");
            // 2.使用ConnectionFactory对象来创建一个Connection对象
            Connection connection = connectionFactory.createConnection();
            // 3.开启连接，需要调用Connection对象的start方法
            connection.start();
            /* * 4.使用Connection对象创建一个Session对象 * 第一个参数：是否开启事务(ActiceMQ的事务)，一般不使用分布式事务，因为它特别消耗性能，而且顾客体验特别差， * 现在互联网的做法是保证数据的最终一致(也就是允许暂时数据不一致)。比如顾客下单购买东西，一旦订单生成完就立刻响应给用户 * 下单成功。至于下单后一系列的操作，比如通知会计记账、通知物流发货、商品数量同步等等都先不用管，只需要发送一条消息到消息队列， * 消息队列来告知各模块进行相应的操作，一次告知不行就两次，直到完成所有相关操作为止，这也就做到了数据的最终一致性。 * 如果第一个参数设为true，那么第二个参数将会被直接忽略掉。如果第一个参数为false，那么第二个参数才有意义。 * 第二个参数：消息的应答模式，常见的有手动应答和自动应答两种模式。我们一般使用自动应答模式。 */
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 5.使用Session对象创建一个Destination对象，目的地有两种形式，一种是queue，一种是topic
            Queue queue = session.createQueue("test-queue"); // 不同名字的Queue相当于是不同的队列，你要是发生消息的时候，你要注意你这个消息要往哪个队列中发送
            // 6.使用Session对象创建一个Producer对象
            MessageProducer producer = session.createProducer(queue);
            // 7.使用Producer对象发生消息
            TextMessage textMessage = new ActiveMQTextMessage();
            textMessage.setText("使用ActiceMQ 发送的队列消息");
            // TextMessage textMessage2 = session.createTextMessage("使用ActiceMQ 发送的队列消息");
            producer.send(textMessage);
            // 8.关闭资源
            producer.close();
            session.close();
            connection.close();
        }
    
    }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22
 *  23
 *  24
 *  25
 *  26
 *  27
 *  28
 *  29
 *  30
 *  31
 *  32
 *  33
 *  34
 *  35
 *  36
 *  37
 *  38
 *  39
 *  40

我们运行上面的测试方法，执行成功后，我们到activemq的后台管理系统，点击”Queues”，可以看到我们刚才发送的那条消息”test-queue”。   
![这里写图片描述][SouthEast 2]   
接着我们点击”test-queue”，会看到如下图所示界面，可以看到刚才发送的那条消息的ID信息，Persistence（持久化）为永久保存，Priorty（优先级）为4，Redelivered（是否重复投递消息）为false，即不重复投递消息。   
![这里写图片描述][SouthEast 3]   
紧接着我们点击ID那一长串字符串，会看到如下图所示界面，我们在”Message Details”一栏，可以看到我们发送的消息信息。   
![这里写图片描述][SouthEast 4]

[SouthEast]: /images/20220605/c09fae1cdc464acaa19d3744d013ecca.png
[SouthEast 1]: /images/20220605/cdec1b0266d54113b5d9c7036e81596f.png
[SouthEast 2]: /images/20220605/d788971073154bac84b5292b207afb2c.png
[SouthEast 3]: /images/20220605/fcdf02c0dc0a4466aab3b85eb597d0ea.png
[SouthEast 4]: /images/20220605/dfa58dead3e249bc931b13276acc038e.png