代码地址 https://github.com/pengbiaobeyond/rabbitmq

    1、之前两节讨论可以总结为对比，virtual host,五种类型（点对点分为普通，公平，发布订阅分为三种，direct,fanout,topic,headers）；

    2、
    消费者如何确保消息一定 能够消费成功？
    通过应答形式的设置，默认是自动应答，无论消费是否正常都会将消息全队列中清除，但是可以更改为手动应答，只要不成功就不应答，则消息就会一直保存在队列中；

    如果RabbitMQ服务器宕机了，消息会丢失吗？
    RabbitMQ服务器支持消息持久化机制，会把消息持久在硬盘上；

    3、mq事务
    问题产生背景:
    生产者发送消息出去之后，不知道到底有没有发送到RabbitMQ服务器，默认是不知道的。而且有的时候我们在发送消息之后，后面的逻辑出问题了，我们不想要发送之前的消息了，需要撤回该怎么做。
    解决方案:
    1.AMQP事务机制
    2.Confirm模式
    事务模式:
    txSelect 将当前channel设置为transaction模式
    txCommit 提交当前事务
    txRollback 事务回滚

    1. public class Producer {
    2.     private static final String QUEUE_NAME = "test_queue";
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         // 1.获取连接
    6.         Connection newConnection = MQConnectionUtils.newConnection();
    7.         // 2.创建通道
    8.         Channel channel = newConnection.createChannel();
    9.         // 3.创建队列声明
    10.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    11.         // 将当前管道设置为 txSelect 将当前channel设置为transaction模式 开启事务
    12.         channel.txSelect();
    13.         String msg = "test_yushengjun110";
    14.         try {
    15.             // 4.发送消息
    16.             channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
    17.             // int i = 1 / 0;
    18.             channel.txCommit();// 提交事务
    19.             System.out.println("生产者发送消息:" + msg);
    20.         } catch (Exception e) {
    21.             System.out.println("消息进行回滚操作");
    22.             channel.txRollback();// 回滚事务
    23.         } finally {
    24.             channel.close();
    25.             newConnection.close();
    26.         }
    28.     }
    30. }

    4、SpringBoot整合RabbitMQ
    生产者：

    1. <parent>
    2.         <groupId>org.springframework.boot</groupId>
    3.         <artifactId>spring-boot-starter-parent</artifactId>
    4.         <version>2.0.0.RELEASE</version>
    5.     </parent>
    6.     <dependencies>
    8.         <!-- springboot-web组件 -->
    9.         <dependency>
    10.             <groupId>org.springframework.boot</groupId>
    11.             <artifactId>spring-boot-starter-web</artifactId>
    12.         </dependency>
    13.         <!-- 添加springboot对amqp的支持 -->
    14.         <dependency>
    15.             <groupId>org.springframework.boot</groupId>
    16.             <artifactId>spring-boot-starter-amqp</artifactId>
    17.         </dependency>
    18.         <dependency>
    19.             <groupId>org.apache.commons</groupId>
    20.             <artifactId>commons-lang3</artifactId>
    21.         </dependency>
    22.         <!--fastjson -->
    23.         <dependency>
    24.             <groupId>com.alibaba</groupId>
    25.             <artifactId>fastjson</artifactId>
    26.             <version>1.2.49</version>
    27.         </dependency>
    28.     </dependencies>
    1. spring:
    2.   rabbitmq:
    3.   ####连接地址
    4.     host: 127.0.0.1
    5.    ####端口号   
    6.     port: 5672
    7.    ####账号 
    8.     username: guest
    9.    ####密码  
    10.     password: guest
    11.    ### 地址
    12.     virtual-host: /


    1. @Component
    2. public class FanoutConfig {
    4.     // 邮件队列
    5.     private String FANOUT_EMAIL_QUEUE = "fanout_eamil_queue";
    7.     // 短信队列
    8.     private String FANOUT_SMS_QUEUE = "fanout_sms_queue";
    9.     // 短信队列
    10.     private String EXCHANGE_NAME = "fanoutExchange";
    12.     // 1.定义队列邮件
    13.     @Bean
    14.     public Queue fanOutEamilQueue() {
    15.         return new Queue(FANOUT_EMAIL_QUEUE);
    16.     }
    18.     @Bean
    19.     public Queue fanOutSmsQueue() {
    20.         return new Queue(FANOUT_SMS_QUEUE);
    21.     }
    23.     // 2.定义交换机
    24.     @Bean
    25.     FanoutExchange fanoutExchange() {
    26.         return new FanoutExchange(EXCHANGE_NAME);
    27.     }
    29.     // 3.队列与交换机绑定邮件队列
    30.     @Bean
    31.     Binding bindingExchangeEamil(Queue fanOutEamilQueue, FanoutExchange fanoutExchange) {
    32.         return BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);
    33.     }
    35.     // 4.队列与交换机绑定短信队列
    36.     @Bean
    37.     Binding bindingExchangeSms(Queue fanOutSmsQueue, FanoutExchange fanoutExchange) {
    38.         return BindingBuilder.bind(fanOutSmsQueue).to(fanoutExchange);
    39.     }
    40. }


    1. @Component
    2. public class FanoutProducer {
    3.     @Autowired
    4.     private AmqpTemplate amqpTemplate;
    6.     public void send(String queueName) {
    7.         String msg = "my_fanout_msg:" + new Date();
    8.         System.out.println(msg + ":" + msg);
    9.         amqpTemplate.convertAndSend(queueName, msg);
    10.     }
    11. }


    2. @RestController
    3. public class ProducerController {
    4.     @Autowired
    5.     private FanoutProducer fanoutProducer;
    7.     @RequestMapping("/sendFanout")
    8.     public String sendFanout(String queueName) {
    9.         fanoutProducer.send(queueName);
    10.         return "success";
    11.     }
    12. }

    消费者：

    1. <parent>
    2.         <groupId>org.springframework.boot</groupId>
    3.         <artifactId>spring-boot-starter-parent</artifactId>
    4.         <version>2.0.0.RELEASE</version>
    5.     </parent>
    6.     <dependencies>
    8.         <!-- springboot-web组件 -->
    9.         <dependency>
    10.             <groupId>org.springframework.boot</groupId>
    11.             <artifactId>spring-boot-starter-web</artifactId>
    12.         </dependency>
    13.         <!-- 添加springboot对amqp的支持 -->
    14.         <dependency>
    15.             <groupId>org.springframework.boot</groupId>
    16.             <artifactId>spring-boot-starter-amqp</artifactId>
    17.         </dependency>
    18.         <dependency>
    19.             <groupId>org.apache.commons</groupId>
    20.             <artifactId>commons-lang3</artifactId>
    21.         </dependency>
    22.         <!--fastjson -->
    23.         <dependency>
    24.             <groupId>com.alibaba</groupId>
    25.             <artifactId>fastjson</artifactId>
    26.             <version>1.2.49</version>
    27.         </dependency>
    28.     </dependencies>


    1. @Component
    2. @RabbitListener(queues = "fanout_eamil_queue")
    3. public class FanoutEamilConsumer {
    4.     @RabbitHandler
    5.     public void process(String msg) throws Exception {
    6.         System.out.println("邮件消费者获取生产者消息msg:" + msg);
    7.     }
    8. }
    11. @Component
    12. @RabbitListener(queues = "fanout_sms_queue")
    13. public class FanoutSmsConsumer {
    14.     @RabbitHandler
    15.     public void process(String msg) {
    16.         System.out.println("短信消费者获取生产者消息msg:" + msg);
    17.     }
    18. }