activeMQ入門+spring boot整合activeMQ
最近想要學習MOM(消息中間件:Message Oriented Middleware),就從比較基礎的activeMQ學起,rabbitMQ、zeroMQ、rocketMQ、Kafka等後續再去學習。
上面說activeMQ是一種消息中間件,可是為什麽要使用activeMQ?
在沒有使用JMS的時候,很多應用會出現同步通信(客戶端發起請求後需要等待服務端返回結果才能繼續執行)、客戶端服務端耦合、單一點對點(P2P)通信的問題,JMS可以通過面向消息中間件的方式很好的解決了上面的問題。
JMS規範術語:
Provider/MessageProvider:生產者
Consumer/MessageConsumer:消費者
消息形式:
1、點對點(queue)
2、一對多(topic)
ConnectionFactory:連接工廠,JMS用它創建連接
Connnection:JMS Client到JMS Provider的連接
Destination:消息目的地,由Session創建
Session:會話,由Connection創建,實質上就是發送、接受消息的一個線程,因此生產者、消費者都是Session創建的
我這裏安裝的是Windows版本的,安裝好了之後就是這樣的目錄
到bin目錄下,啟動activemq.bat
這樣就啟動成功了。
訪問http://localhost:8161/admin/index.jsp可以看到管控臺,如下圖:
spring boot整合activeMQ:
pom.xml中引用
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-activemq</artifactId> </dependency> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool</artifactId> <!-- <version>5.7.0</version> --> </dependency>
在application.properties中配置activeMQ連接:
spring.activemq.broker-url=tcp://localhost:61616 spring.activemq.in-memory=true spring.activemq.pool.enabled=false
創建消息生產者:
/** * @author huangzhang * @description * @date Created in 2018/7/16 18:57 */ @Service("provider") public class Provider { @Autowired private JmsMessagingTemplate jmsMessagingTemplate; public void sendMessage(Destination destination, final String message){ jmsMessagingTemplate.convertAndSend(destination,message); }
//消費消費者返回的隊列"return.queue"中的消息 @JmsListener(destination="return.queue") public void consumerMessage(String string){ System.out.println("從return.queue隊列收到的回復報文為:"+string); } }
創建第一個消費者:
/** * @author huangzhang * @description * @date Created in 2018/7/16 19:31 */ @Component public class Consumer { @JmsListener(destination = "mytest.queue") public void receiveQueue(String text){ System.out.println("Consumer收到的報文為:"+text); } }
創建第二個消費者(這裏不光消費了生產者插入隊列中的message,而且將返回值插入到了"return.queue"隊列中):
/** * @author huangzhang * @description * @date Created in 2018/7/16 19:33 */ @Component public class Consumer1 { @JmsListener(destination = "mytest.queue") @SendTo("return.queue") public String receiveQueue(String message){ System.out.println("Consumer1收到的報文為:"+message); return "========return message "+message; } }
測試方法:
@Service public class SpringbootJmsApplicationTests { @Autowired private Provider provider; public void contextLoads() throws InterruptedException { Destination destination = new ActiveMQQueue("mytest.queue"); for(int i=0; i<10; i++){ provider.sendMessage(destination, "huangzhang "+i); } } }
這裏我在controller中調用了測試方法:
/** * @author huangzhang * @description * @date Created in 2018/7/16 20:23 */ @Controller public class Test { @Autowired SpringbootJmsApplicationTests springbootJmsApplicationTests; @RequestMapping("/") @ResponseBody public String test01()throws Exception{ springbootJmsApplicationTests.contextLoads(); return "success!"; } }
訪問http://localhost:8080/對此demo進行測試
這裏是執行結果,可以看出,兩個消費者分別消費了生產者放入消息隊列中的消息,並且Consumer1消費者將返回結果放入了隊列中供生產者消費。
查看Queues
這裏可以看出我們生產者循環往mytest.queue隊列中寫入10次,由兩個消費者消費了這10次消息
consumer1消費了5次消息,並往返回隊列return.queue中寫入5次,由原生產者消費了者5次消息
到這裏一個簡單的spring boot整合activeMQ的demo就完成了(請多指正)。
activeMQ入門+spring boot整合activeMQ