我們使用jms一般是使用spring-jms和activemq相結(jié)合，通過(guò)spring Boot為我們配置好的JmsTemplate發(fā)送消息到指定的目的地Destination。本文以點(diǎn)到點(diǎn)消息模式為例，演示如何在Spring Boot中整合 JMS 和 Active MQ ，實(shí)現(xiàn) MQ 消息的生產(chǎn)與消費(fèi)。

點(diǎn)到點(diǎn)消息模式定義：當(dāng)消息發(fā)送者發(fā)送消息，消息代理獲得消息后，把消息放入一個(gè)隊(duì)列里，當(dāng)有消息接收者來(lái)接收消息的時(shí)候，消息將從隊(duì)列里取出并且傳遞給接收者，這時(shí)候隊(duì)列里就沒(méi)有此消息了。隊(duì)列Queue的存在使得消息的異步傳輸成為可能。

1 安裝ActiveMQ

關(guān)于具體的安裝步驟，請(qǐng)參考專門介紹安裝與部署的另一篇博文《ActiveMQ安裝及部署教程圖解》。

2 配置Active MQ 依賴

關(guān)于如何搭建 Spring Boot工程，請(qǐng)移步《如何使用intellij IDEA搭建Spring Boot項(xiàng)目》。假設(shè)項(xiàng)目已經(jīng)創(chuàng)建完畢，在項(xiàng)目 pom 文件中加入Active MQ依賴：

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-activemq</artifactId>
 <version>2.2.7.RELEASE</version>
</dependency>

3 修改application.properties配置文件

在application.properties中配置Active MQ：

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false
#默認(rèn)值false，表示點(diǎn)到點(diǎn)模式，true時(shí)代表發(fā)布訂閱模式
spring.jms.pub-sub-domain=false

4 創(chuàng)建消息生產(chǎn)者和消費(fèi)者

本文以點(diǎn)到點(diǎn)消息模式演示。新增消息生產(chǎn)者：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.Destination;

@Service("producer")
public class Producer {
 /**
  * 也可以注入JmsTemplate，JmsMessagingTemplate對(duì)JmsTemplate進(jìn)行了封裝
  */
 @Autowired
 private JmsMessagingTemplate jmsTemplate;

 /**
  * 發(fā)送消息，destination是發(fā)送到的隊(duì)列，message是待發(fā)送的消息
  *
  * @param destination
  * @param message
  */
 public void sendMessage(Destination destination, final String message) {
  jmsTemplate.convertAndSend(destination, message);
 }
}

通過(guò)JmsMessagingTemplate 的convertAndSend方法向目的地 destination發(fā)送消息。 新增消息監(jiān)聽(tīng)者：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class ConsumerListener {
 private static Logger logger = LoggerFactory.getLogger(Consumer.class);
 /**
  * 使用JmsListener配置消費(fèi)者監(jiān)聽(tīng)的隊(duì)列
  * @param receivedMsg 接收到的消息
  */
 @JmsListener(destination = "myDest.queue")
 public void receiveQueue(String receivedMsg) {
  logger.info("Consumer收到的報(bào)文為: {}", receivedMsg);
 }
}

@JmsListener是Spring 4.1 提供的一個(gè)新特性，用于簡(jiǎn)化JMS開(kāi)發(fā)，只需使用此注解的屬性destination指定要監(jiān)聽(tīng)的目的地【myDest.queue】，即可接收該目的地發(fā)送的消息。消費(fèi)者2 ConsumerListener2 的代碼同上，但是日志如下：

logger.info("Consumer2收到的報(bào)文為: {}", receivedMsg);

溫馨提示，消息消費(fèi)者和生產(chǎn)者的類上必須加上注解 @Component 或者 @Service，使得消息消費(fèi)者類就會(huì)被委派給Listener類，原理類似于使用SessionAwareMessageListener以及MessageListenerAdapter來(lái)實(shí)現(xiàn)消息驅(qū)動(dòng)POJO。

5 測(cè)試 MQ

在Junit測(cè)試類中新增測(cè)試方法

import org.apache.activemq.command.ActiveMQQueue;
import javax.jms.Destination;

@Autowired
private Producer producer;
@Test
public void jmsActiveMqTest() throws InterruptedException {
 Destination destination = new ActiveMQQueue("myDest.queue");
 for (int i = 0; i < 21; i++) {
  producer.sendMessage(destination, String.format("My name is Wiener%s", i));
 }
}

控制臺(tái)顯示消費(fèi)者接收到的消息如下：

Consumer收到的報(bào)文為: My name is Wiener0
Consumer2 收到的報(bào)文為：My name is Wiener1
Consumer收到的報(bào)文為: My name is Wiener2
Consumer2 收到的報(bào)文為：My name is Wiener3
Consumer收到的報(bào)文為: My name is Wiener4
Consumer2 收到的報(bào)文為：My name is Wiener5
Consumer收到的報(bào)文為: My name is Wiener6
Consumer2 收到的報(bào)文為：My name is Wiener7
...

經(jīng)過(guò)上面的五個(gè)步驟，spring Boot、Jms 和 Active MQ 就基本上整合完成了，是不是使用起來(lái)很方便了！

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論