原文:https://www.jianshu.com/p/e1258c004314css
RabbitMQ需啊erlang環境,因此首先安裝對應版本的erlang,可在RabbitMQ官網下載java
# rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm
使用yum安裝RabbitMQ,避免缺乏依賴包引發的安裝失敗web
# yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm
啓動RabbitMQspring
# /sbin/service rabbitmq-server start
因爲RabbitMQ默認提供的guest用戶只能本地訪問,因此額外建立用戶用於測試centos
# /sbin/rabbitmqctl add_user test test123 用戶名:test,密碼:test123
開啓web管理插件bash
# rabbitmq-plugins enable rabbitmq_management
並使用以前建立的用戶登陸,並設置該用戶爲administrator,虛擬主機地址爲/併發
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> </dependencies>
spring.rabbitmq.host=192.168.1.107
spring.rabbitmq.port=5672
spring.rabbitmq.username=test spring.rabbitmq.password=test123 spring.rabbitmq.publisher-confirms=true spring.rabbitmq.publisher-returns=true spring.rabbitmq.template.mandatory=true
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitConfig { //聲明隊列 @Bean public Queue queue1() { return new Queue("hello.queue1", true); // true表示持久化該隊列 } @Bean public Queue queue2() { return new Queue("hello.queue2", true); } //聲明交互器 @Bean TopicExchange topicExchange() { return new TopicExchange("topicExchange"); } //綁定 @Bean public Binding binding1() { return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1"); } @Bean public Binding binding2() { return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#"); } }
共聲明瞭2個隊列,分別是hello.queue1,hello.queue2,交換器類型爲TopicExchange,並與hello.queue1,hello.queue2隊列分別綁定。app
import java.util.UUID; import javax.annotation.PostConstruct; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback { @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init() { rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnCallback(this); } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { System.out.println("消息發送成功:" + correlationData); } else { System.out.println("消息發送失敗:" + cause); } } @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println(message.getMessageProperties().getCorrelationIdString() + " 發送失敗"); } //發送消息,不須要實現任何接口,供外部調用。 public void send(String msg){ CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString()); System.out.println("開始發送消息 : " + msg.toLowerCase()); String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString(); System.out.println("結束髮送消息 : " + msg.toLowerCase()); System.out.println("消費者響應 : " + response + " 消息處理完成"); } }
要點:dom
1.注入RabbitTemplateide
2.實現RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口(非必須)。
ConfirmCallback接口用於實現消息發送到RabbitMQ交換器後接收ack回調。ReturnCallback接口用於實現消息發送到RabbitMQ交換器,但無相應隊列與交換器綁定時的回調。
3.實現消息發送方法。調用rabbitTemplate相應的方法便可。rabbitTemplate經常使用發送方法有
rabbitTemplate.send(message); //發消息,參數類型爲org.springframework.amqp.core.Message rabbitTemplate.convertAndSend(object); //轉換併發送消息。 將參數對象轉換爲org.springframework.amqp.core.Message後發送 rabbitTemplate.convertSendAndReceive(message) //轉換併發送消息,且等待消息者返回響應消息。
針對業務場景選擇合適的消息發送方式便可。
spring.rabbitmq.host=192.168.1.107 spring.rabbitmq.port=5672 spring.rabbitmq.username=test spring.rabbitmq.password=test123 spring.rabbitmq.listener.concurrency=2 //最小消息監聽線程數 spring.rabbitmq.listener.max-concurrency=2 //最大消息監聽線程數
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class Receiver { @RabbitListener(queues = "hello.queue1") public String processMessage1(String msg) { System.out.println(Thread.currentThread().getName() + " 接收到來自hello.queue1隊列的消息:" + msg); return msg.toUpperCase(); } @RabbitListener(queues = "hello.queue2") public void processMessage2(String msg) { System.out.println(Thread.currentThread().getName() + " 接收到來自hello.queue2隊列的消息:" + msg); } }
因爲定義了2個隊列,因此分別定義不一樣的監聽器監聽不一樣的隊列。因爲最小消息監聽線程數和最大消息監聽線程數都是2,因此每一個監聽器各有2個線程實現監聽功能。
要點:
1.監聽器參數類型與消息實際類型匹配。在生產者中發送的消息實際類型是String,因此這裏監聽器參數類型也是String。
2.若是監聽器須要有響應返回給生產者,直接在監聽方法中return便可。
import java.util.Date; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.sam.demo.rabbitmq.Application; import com.sam.demo.rabbitmq.sender.Sender; @RunWith(value=SpringJUnit4ClassRunner.class) @SpringBootTest(classes = Application.class) public class RabbitTests { @Autowired private Sender sender; @Test public void sendTest() throws Exception { while(true){ String msg = new Date().toString(); sender.send(msg); Thread.sleep(1000); } } }
輸出:
開始發送消息 : wed mar 29 23:20:52 cst 2017 SimpleAsyncTaskExecutor-1 接收到來自hello.queue2隊列的消息:Wed Mar 29 23:20:52 CST 2017 SimpleAsyncTaskExecutor-2 接收到來自hello.queue1隊列的消息:Wed Mar 29 23:20:52 CST 2017 結束髮送消息 : wed mar 29 23:20:52 cst 2017 消費者響應 : WED MAR 29 23:20:52 CST 2017 消息處理完成 ------------------------------------------------ 消息發送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886] 開始發送消息 : wed mar 29 23:20:53 cst 2017 SimpleAsyncTaskExecutor-1 接收到來自hello.queue1隊列的消息:Wed Mar 29 23:20:53 CST 2017 SimpleAsyncTaskExecutor-2 接收到來自hello.queue2隊列的消息:Wed Mar 29 23:20:53 CST 2017 結束髮送消息 : wed mar 29 23:20:53 cst 2017 消費者響應 : WED MAR 29 23:20:53 CST 2017 消息處理完成 ------------------------------------------------ 消息發送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34] 開始發送消息 : wed mar 29 23:20:54 cst 2017 SimpleAsyncTaskExecutor-2 接收到來自hello.queue1隊列的消息:Wed Mar 29 23:20:54 CST 2017 SimpleAsyncTaskExecutor-1 接收到來自hello.queue2隊列的消息:Wed Mar 29 23:20:54 CST 2017 結束髮送消息 : wed mar 29 23:20:54 cst 2017 消費者響應 : WED MAR 29 23:20:54 CST 2017 消息處理完成 ------------------------------------------------
經常使用交換器類型以下:
Direct(DirectExchange):direct 類型的行爲是"先匹配, 再投送". 即在綁定時設定一個 routing_key, 消息的routing_key徹底匹配時, 纔會被交換器投送到綁定的隊列中去。
Topic(TopicExchange):按規則轉發消息(最靈活)。
Headers(HeadersExchange):設置header attribute參數類型的交換機。
Fanout(FanoutExchange):轉發消息到全部綁定隊列。