訊息佇列RabbitMQ與Spring整合
阿新 • • 發佈:2018-12-31
1.RabbitMQ簡介
RabbitMQ是流行的開源訊息佇列系統,用erlang語言開發。RabbitMQ是AMQP(高階訊息佇列協議)的標準實現。
官網:http://www.rabbitmq.com/
2.Spring整合RabbitMQ
2.1 maven配置
//pom.xml <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.5.1</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.4.5.RELEASE</version> </dependency>
2.2 rabbmitmq配置檔案
//rabbitmq-config.properties
mq.host=127.0.0.1
mq.username=test
mq.password=123456
mq.port=5672
mq.vhost=testmq
2.3 Spring配置
//application-mq.xml <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" > <description>rabbitmq 連線服務配置</description> <!-- 連線配置 --> <rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}" virtual-host="${mq.vhost}"/> <rabbit:admin connection-factory="connectionFactory"/> <!-- spring template宣告--> <rabbit:template exchange="amqpExchange" id="amqpTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" /> <!-- 訊息物件json轉換類 --> <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> </beans>
3. 在Spring中使用RabbitMQ
3.1 申明一個訊息佇列Queue
//application-mq.xml
<rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false" />
說明:
durable:是否持久化
exclusive: 僅建立者可以使用的私有佇列,斷開後自動刪除
auto_delete: 當所有消費客戶端連線斷開後,是否自動刪除佇列
3.2 交換機定義
//application-mq.xml
<rabbit:direct-exchange name="test-mq-exchange" durable="true" auto-delete="false" id="test-mq-exchange">
<rabbit:bindings>
<rabbit:binding queue="test_queue_key" key="test_queue_key"/>
</rabbit:bindings>
</rabbit:direct-exchange>
說明:
rabbit:direct-exchange:定義exchange模式為direct,意思就是訊息與一個特定的路由鍵完全匹配,才會轉發。
rabbit:binding:設定訊息queue匹配的key
3.3 傳送訊息Producer
//MQProducer.java
public interface MQProducer {
/**
* 傳送訊息到指定佇列
* @param queueKey
* @param object
*/
public void sendDataToQueue(String queueKey, Object object);
}
@Service
public class MQProducerImpl implements MQProducer {
@Autowired
private AmqpTemplate amqpTemplate;
private final static Logger LOGGER = Logger.getLogger(MQProducerImpl.class);
/* (non-Javadoc)
* @see com.stnts.tita.rm.api.mq.MQProducer#sendDataToQueue(java.lang.String, java.lang.Object)
*/
@Override
public void sendDataToQueue(String queueKey, Object object) {
try {
amqpTemplate.convertAndSend(queueKey, object);
} catch (Exception e) {
LOGGER.error(e);
}
}
}
說明:
convertAndSend:將Java物件轉換為訊息傳送到匹配Key的交換機中Exchange,由於配置了JSON轉換,這裡是將Java物件轉換成JSON字串的形式。原文:Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.
3.4 非同步接收訊息Consumer
定義監聽器
//QueueListenter.java
@Component
public class QueueListenter implements MessageListener {
@Override
public void onMessage(Message msg) {
try{
System.out.print(msg.toString());
}catch(Exception e){
e.printStackTrace();
}
}
}
監聽配置
//application-mq.xml
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
<rabbit:listener queues="test_queue_key" ref="queueListenter"/>
</rabbit:listener-container>
說明:
queues:監聽的佇列,多個的話用逗號(,)分隔
ref:監聽器
3.5 JUnit測試
//TestQueue.java
@RunWith(value = SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {
"classpath:/ApplicationContext/ApplicationContext-mq.xml"})
public class TestQueue{
@Autowired
MQProducer mqProducer;
final String queue_key = "test_queue_key";
@Test
public void send(){
Map<String,Object> msg = new HashMap()<>;
msg.put("data","hello,rabbmitmq!");
mqProducer.sendDataToQueue(query_key,msg);
}
}
執行測試程式,Run with JUnit,會發送一條訊息到test_queue,監聽器監聽到訊息後,打印出訊息。
至此,已經完成了Spring和RabbmitMQ整合,配置,和使用。
個人技術分享微信公眾號,歡迎關注一起交流