<!--rabbitmq-->
 	 <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>

server:
  port: 8090
spring:
  application:
    name: rabbitmq-provider
  rabbitmq:
    host: 192.168.152.173
    port: 5672     #注意15672是web頁面端口，消息發(fā)送端口為5672
    username: guest
    password: guest

@Configuration
public class DirectRabbitConfig {

    /**
     * 創(chuàng)建消息隊(duì)列 起名：TestDirectQueue
     * durable:是否持久化,默認(rèn)是false,持久化隊(duì)列：會被存儲在磁盤上，當(dāng)消息代理重啟時(shí)仍然存在，暫存隊(duì)列：當(dāng)前連接有效
     * exclusive:默認(rèn)也是false，只能被當(dāng)前創(chuàng)建的連接使用，而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級高于durable
     * autoDelete:是否自動刪除，當(dāng)沒有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列，該隊(duì)列會自動刪除。
     * 一般設(shè)置一下隊(duì)列的持久化就好,其余兩個(gè)就是默認(rèn)false
     * @return new Queue(name,durable,exclusive,autoDelete)
     */
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }

    /**
     * Direct交換機(jī) 起名：TestDirectExchange
     */
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }

    /**
     * 綁定 
     * 將隊(duì)列和交換機(jī)綁定, 并設(shè)置用于匹配鍵（路由鍵）：TestDirectRouting
     */
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }   
}

@RestController
@RequestMapping("/rabbit")
public class SendMessageController {

    @Autowired
    private RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,這提供了接收/發(fā)送等等方法

    @RequestMapping("/send")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "League of Legends never dies！";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //將消息攜帶綁定鍵值：TestDirectRouting 發(fā)送到交換機(jī)TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

@Component
@RabbitListener(queues = "TestDirectQueue")//監(jiān)聽的隊(duì)列名稱 TestDirectQueue
public class DirectReceiver01 {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver01消費(fèi)者收到的消息: " + testMessage.toString());
    }
}

@Configuration
public class TopicRabbitConfig {
    //綁定鍵
      public final static String man = "topic.man";
      public final static String woman = "topic.woman";

      @Bean
      public Queue firstQueue() {
          return new Queue(TopicRabbitConfig.man);
      }

      @Bean
      public Queue secondQueue() {
          return new Queue(TopicRabbitConfig.woman);
      }

      @Bean
      TopicExchange exchange() {
          return new TopicExchange("topicExchange");
      }


      //將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man
      //這樣只要是消息攜帶的路由鍵是topic.man,才會分發(fā)到該隊(duì)列
      @Bean
      Binding bindingExchangeMessage() {
          return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
      }

      //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
      // 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊(duì)列
      @Bean
      Binding bindingExchangeMessage2() {
          return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
      }
}

@RequestMapping("rabbit")
@RestController
public class TopicSendController {
    @Autowired
    private RabbitTemplate template;

    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: M A N ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> manMap = new HashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        template.convertAndSend("topicExchange", "topic.man", manMap);
        return "ok";
    }

    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: woman is all ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));

        Map<String, Object> womanMap = new HashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        template.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消費(fèi)者收到消息  : " + testMessage.toString());
    }
}

@Component
@RabbitListener(queues = "topic.woman")
public class TopicWoManReceiver {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicWoManReceiver消費(fèi)者收到消息  : " + testMessage.toString());
    }
}

@Configuration
public class FanoutRabbitConfig {

    /**
     * 創(chuàng)建三個(gè)隊(duì)列 ：fanout.A   fanout.B  fanout.C
     * 將三個(gè)隊(duì)列都綁定在交換機(jī) fanoutExchange 上
     * 因?yàn)槭巧刃徒粨Q機(jī), 路由鍵無需配置,配置也不起作用
     */

    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

@RestController
@RequestMapping("rabbit")
public class FanoutController {

    @Autowired
    private RabbitTemplate template;

    @GetMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: testFanoutMessage ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        template.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }
}

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費(fèi)者收到消息  : " +testMessage.toString());
    }
}

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB{
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費(fèi)者收到消息  : " +testMessage.toString());
    }
}

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消費(fèi)者收到消息  : " +testMessage.toString());
    }
}

server:
  port: 8090
spring:
  application:
    name: rabbitmq-consumer
  rabbitmq:
    host: 192.168.152.173
    port: 5672
    username: guest
    password: guest
    #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)
    publisher-confirm-type: correlated
    #確認(rèn)消息已發(fā)送到隊(duì)列(Queue)
    publisher-returns: true

/**
 * 配置相關(guān)的消息確認(rèn)回調(diào)函數(shù)
 */
@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //設(shè)置開啟Mandatory,才能觸發(fā)回調(diào)函數(shù),無論消息推送結(jié)果怎么樣都強(qiáng)制調(diào)用回調(diào)函數(shù)
        rabbitTemplate.setMandatory(true);

        //確認(rèn)回調(diào)
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean condition, String cause) {
                System.out.println("ConfirmCallback:     " + "相關(guān)數(shù)據(jù)：" + correlationData);
                System.out.println("ConfirmCallback:     " + "確認(rèn)情況：" + condition);
                System.out.println("ConfirmCallback:     " + "原因：" + cause);
            }
        });

        //設(shè)置返回消息的回調(diào)
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     " + "消息：" + message);
                System.out.println("ReturnCallback:     " + "回應(yīng)碼：" + replyCode);
                System.out.println("ReturnCallback:     " + "回應(yīng)信息：" + replyText);
                System.out.println("ReturnCallback:     " + "交換機(jī)：" + exchange);
                System.out.println("ReturnCallback:     " + "路由鍵：" + routingKey);
            }
        });
        return rabbitTemplate;
    }
}

 @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //non-existent-exchange這個(gè)交換機(jī)沒有創(chuàng)建和配置
        template.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

 @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

 @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

rabbitmq:
  host: 192.168.152.173
  port: 5672
  username: guest
  password: guest
  listener:
    direct:
      acknowledge-mode: auto   # manual：手動

package com.zsn.demo.config;

import com.zsn.demo.resceiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author： zhouzhou
 * @date：2021/7/22 9:49
 */
@Configuration
public class MessageListenerConfig {
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收處理類

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認(rèn)是自動確認(rèn)，這里改為手動確認(rèn)消息
        //設(shè)置一個(gè)隊(duì)列
        container.setQueueNames("TestDirectQueue");
        //如果同時(shí)設(shè)置多個(gè)如下： 前提是隊(duì)列都是必須已經(jīng)創(chuàng)建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");


        //另一種設(shè)置隊(duì)列的方法,如果使用這種情況,那么要設(shè)置多個(gè),就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);

        return container;
    }
}

package com.zsn.demo.resceiver;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @author： zhouzhou
 * @date：2021/7/22 9:50
 */
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
    /**
     * 用于處理接收到的 Rabbit 消息的回調(diào)。
     * 實(shí)現(xiàn)者應(yīng)該處理指定的信息
     * 通常通過給定的會話發(fā)送回復(fù)消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            System.out.println("消費(fèi)的主題消息來自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個(gè)參數(shù)，手動確認(rèn)可以被批處理，當(dāng)該參數(shù)為 true 時(shí)，則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個(gè)參數(shù)，true會重新放回隊(duì)列，所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時(shí)候使用拒絕
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

    //{key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private Map<String, String> mapStringToMap(String str,int entryNum ) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
   /**
     * 用于處理接收到的 Rabbit 消息的回調(diào)。
     * 實(shí)現(xiàn)者應(yīng)該處理指定的信息
     * 通常通過給定的會話發(fā)送回復(fù)消息
     * @param message the received AMQP message (never <code>null</code>)
     * @param channel the underlying Rabbit Channel (never <code>null</code>)
     * @throws Exception Any.
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費(fèi)的消息來自的隊(duì)列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費(fèi)到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務(wù)處理流程......");
                
            }
 
            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消費(fèi)的消息來自的隊(duì)列名為："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消費(fèi)到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("執(zhí)行fanout.A中的消息的業(yè)務(wù)處理流程......");
 
            }
            
            channel.basicAck(deliveryTag, true);
//			channel.basicReject(deliveryTag, true);//為true會重新放回隊(duì)列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
    //{key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private Map<String, String> mapStringToMap(String str,int enNum) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",enNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

public interface RabbitConsumerService {
    void process(Channel channel, Message message) throws IOException;
}

@Component
public class RabbitConsumerServiceImpl implements RabbitConsumerService{
    @RabbitHandler
    @RabbitListener(queues = "TestDirectQueue")
    @Override
    public void process(Channel channel, Message message) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            System.out.println("消費(fèi)的主題消息來自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二個(gè)參數(shù)，手動確認(rèn)可以被批處理，當(dāng)該參數(shù)為 true 時(shí)，則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二個(gè)參數(shù)，true會重新放回隊(duì)列，所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時(shí)候使用拒絕
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
    //{key=value,key=value,key=value} 格式轉(zhuǎn)換成map
    private Map<String, String> mapStringToMap(String str,int entryNum ) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

rabbitmq:
    addresses: 192.168.152.193:5672
    username: guest
    password: guest  
    #虛擬host 可以不設(shè)置,使用server默認(rèn)host
    virtual-host: /
    listener:
      simple:
        prefetch: 1               #設(shè)置一次處理一個(gè)消息
        acknowledge-mode: manual  #設(shè)置消費(fèi)端手動 ack
        concurrency: 3            #設(shè)置同時(shí)有3個(gè)消費(fèi)者消費(fèi)
    connection-timeout: 500
    # 確認(rèn)消息是否正確到達(dá)queue,如果沒有則觸發(fā)，如果有則不觸發(fā)
    publisher-returns: true
    #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange)
    publisher-confirm-type: correlated
    template:
      #設(shè)置為 true后 消費(fèi)者在消息沒有被路由到合適隊(duì)列情況下會被return監(jiān)聽，而不會自動刪除
      mandatory: true