09---------RabbitMq的四種交換機模式
首先要匯入依賴
<!-- 新增rabbitmq依賴 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
接著配置相關配置檔案
#rabbitmq
spring.rabbitmq.host=自己的主機或者本地(localhost)
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=/
#\u6D88\u8D39\u8005\u6570\u91CF
spring.rabbitmq.listener.simple.concurrency= 10
spring.rabbitmq.listener.simple.max-concurrency= 10
#\u6D88\u8D39\u8005\u6BCF\u6B21\u4ECE\u961F\u5217\u83B7\u53D6\u7684\u6D88\u606F\u6570\u91CF
spring.rabbitmq.listener.simple.prefetch= 1
#\u6D88\u8D39\u8005\u81EA\u52A8\u542F\u52A8
spring.rabbitmq.listener.simple.auto-startup=true
#\u6D88\u8D39\u5931\u8D25\uFF0C\u81EA\u52A8\u91CD\u65B0\u5165\u961F
spring.rabbitmq.listener.simple.default-requeue-rejected= true
#\u542F\u7528\u53D1\u9001\u91CD\u8BD5
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0
1.使用MQConfig
package MiaoSha.rabbitmq; import java.util.HashMap; import java.util.Map; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.HeadersExchange; 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 MQConfig { public static final String MIAOSHA_QUEUE = "miaosha.queue"; public static final String QUEUE = "queue"; public static final String TOPIC_QUEUE1 = "topic.queue1"; public static final String TOPIC_QUEUE2 = "topic.queue2"; public static final String HEADER_QUEUE = "header.queue"; public static final String TOPIC_EXCHANGE = "topicExchage"; public static final String FANOUT_EXCHANGE = "fanoutxchage"; public static final String HEADERS_EXCHANGE = "headersExchage"; /** * Direct模式 交換機Exchange * */ @Bean public Queue queue() { return new Queue(QUEUE, true); } /** * Topic模式 交換機Exchange * */ @Bean public Queue topicQueue1() { return new Queue(TOPIC_QUEUE1, true); } @Bean public Queue topicQueue2() { return new Queue(TOPIC_QUEUE2, true); } @Bean public TopicExchange topicExchage(){ return new TopicExchange(TOPIC_EXCHANGE); } @Bean public Binding topicBinding1() { return BindingBuilder.bind(topicQueue1()).to(topicExchage()).with("topic.key1"); } @Bean public Binding topicBinding2() { return BindingBuilder.bind(topicQueue2()).to(topicExchage()).with("topic.#"); } /** * Fanout模式 交換機Exchange * */ @Bean public FanoutExchange fanoutExchage(){ return new FanoutExchange(FANOUT_EXCHANGE); } @Bean public Binding FanoutBinding1() { return BindingBuilder.bind(topicQueue1()).to(fanoutExchage()); } @Bean public Binding FanoutBinding2() { return BindingBuilder.bind(topicQueue2()).to(fanoutExchage()); } /** * Header模式 交換機Exchange * */ @Bean public HeadersExchange headersExchage(){ return new HeadersExchange(HEADERS_EXCHANGE); } @Bean public Queue headerQueue1() { return new Queue(HEADER_QUEUE, true); } @Bean public Binding headerBinding() { Map<String, Object> map = new HashMap<String, Object>(); map.put("header1", "value1"); map.put("header2", "value2"); return BindingBuilder.bind(headerQueue1()).to(headersExchage()).whereAll(map).match(); } }
2.MQReceiver
package MiaoSha.rabbitmq; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Service; @Service public class MQReceiver { @RabbitListener(queues=MQConfig.QUEUE) public void receive(String message) { log.info("receive message:"+message); } @RabbitListener(queues=MQConfig.TOPIC_QUEUE1) public void receiveTopic1(String message) { log.info(" topic queue1 message:"+message); } @RabbitListener(queues=MQConfig.TOPIC_QUEUE2) public void receiveTopic2(String message) { log.info(" topic queue2 message:"+message); } @RabbitListener(queues=MQConfig.HEADER_QUEUE) public void receiveHeaderQueue(byte[] message) { log.info(" header queue message:"+new String(message)); } }
3.MQSender
public void send(Object message) {
String msg = RedisService.beanToString(message);
log.info("send message:"+msg);
amqpTemplate.convertAndSend(MQConfig.QUEUE, msg);
}
public void sendTopic(Object message) {
String msg = RedisService.beanToString(message);
log.info("send topic message:"+msg);
amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key1", msg+"1");
amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key2", msg+"2");
}
public void sendFanout(Object message) {
String msg = RedisService.beanToString(message);
log.info("send fanout message:"+msg);
amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE, "", msg);
}
public void sendHeader(Object message) {
String msg = RedisService.beanToString(message);
log.info("send fanout message:"+msg);
MessageProperties properties = new MessageProperties();
properties.setHeader("header1", "value1");
properties.setHeader("header2", "value2");
Message obj = new Message(msg.getBytes(), properties);
amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE, "", obj);
}
4.controller呼叫
@RequestMapping("/mq/header")
@ResponseBody
public Result<String> header() {
sender.sendHeader("hello,imooc");
return Result.success("Hello,world");
}
@RequestMapping("/mq/fanout")
@ResponseBody
public Result<String> fanout() {
sender.sendFanout("hello,imooc");
return Result.success("Hello,world");
}
@RequestMapping("/mq/topic")
@ResponseBody
public Result<String> topic() {
sender.sendTopic("hello,imooc");
return Result.success("Hello,world");
}
@RequestMapping("/mq")
@ResponseBody
public Result<String> mq() {
sender.send("hello,imooc");
return Result.success("Hello,world");
}