【消息队列】消息中间件RabbitMQ急速入门2

Spring 框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的 POJO的消息监听等.
提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象，最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发.
总之就是提高我们的框架整合消息队列的效率,SpringBoot为更方便开发RabbitMQ推出了starter.

<!-- 代码库 -->
    <repositories>
        <repository>
            <id>maven-ali</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public//</url>
            <releases>
                <enabled>true</enabled>
            </releases>
            <snapshots>
                <enabled>true</enabled>
                <updatePolicy>always</updatePolicy>
                <checksumPolicy>fail</checksumPolicy>
            </snapshots>
        </repository>
    </repositories>
    <pluginRepositories>
        <pluginRepository>
            <id>public</id>
            <name>aliyun nexus</name>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
            <releases>
                <enabled>true</enabled>
            </releases>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </pluginRepository>
    </pluginRepositories>

 <!--引入AMQP-->
<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

#消息队列
spring: 
  rabbitmq:
    host: 10.211.63.14
    port: 5672
    virtual-host: /dev
    password: 123456
    username: admin

@Configuration
public class RabbitMQConfig{
    public static final String EXCHANGE_NAME = "exchange_order";
    public static final String QUEUE = "order_queue";
    /**
     * 交换机
     * @return
     */
    @Bean
    public Exchange orderExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(ture).build();
    }
    /**
     * 队列
     * @return
     */
    @Bean
    public Queue orderQueue(){
        return QueueBuilder.durable(QUEUE).build();
    }
    /**
     * 交换机和队列绑定关系
     */
    @Bean
    public Binding orderBinding(Queue queue,Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs();
    }
}

rabbitTemplate.converAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.new","新订单");

@Component
@RabbitMQListener(queue = "order_queue")
public class OrderMQListener{
    /**
     * RabbitHandler 会自动匹配 消息类型（消息自动确认）
     * @param msg
     * @param message
     * @throws IOException
     */
    @RabbitHandler
    public void messageHandler(String body,Message message){
        long msgTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("msgTag="+msgTag);
        System.out.println("message="+message.toString());
        System.out.println("监听到消息：消息内容:"+message.getBody());
    }
}

#旧版，确认消息发送成功，通过实现ConfirmCallBack接口，消息发送到交换机Exchange后触发回调
spring.rabbitmq.publisher-confirms=true
#新版，NONE值是禁用发布确认模式，是默认值，CORRELATED值事发布消息成功到交换机后触发回调方法
spring.rabbitmq.publisher-confirm-type: correlated

@Test
    void testConfirmCallback(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * @param correlationData 配置
             * @param ack 交换机是否收到消息，true是成功，false是失败
             * @param cause 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback======>");
                System.out.println("correlationData======>"+correlationData);
                System.out.println("ack======>"+ack);
                System.out.println("cause======>"+cause);
                if(ack){
                    System.out.println("发送成功");
                    //更新数据库的状态，状态为成功
                }else {
                    System.out.println("发送失败，记录到日志或者数据库");
                    //更新数据库的状态，状态为失败
                }
            }
        });
        //数据库新增一个消息记录，状态是发送，发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.new", "中台收到一条新订单");
    }

#配置为true，则交换机处理消息到路由失败，会返回给生产者
spring.rabbitmq.template.mandatory=true
#或者在temlate对象上设置
template.setMandatory(true);

#新版
spring.rabbitmq.publisher-returns=true

#为true，则交换机处理消息到路由失败会返回给生产者
spring.rabbitmq.template.mandatory=true

    @Test
    void testReturnCallback(){
    //publisher-returns为true则交换机处理消息到路由失败，返回给生产者
        //mandatory为true则消息未被路由到任何一个queue，则回退一条消息给生产者
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                System.out.println("ReturnedMessage:" + returned.toString());
                int replyCode = returned.getReplyCode();
                System.out.println("_______________________");
                System.out.println("replyCode:" + replyCode);
            }
        });
        //数据库新增一个消息记录，状态是发送，发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.new", "中台收到一条新订单");
    }
}

spring:
  rabbitmq: 
    #开启手动确认消息，如果消息重新入队，进行重试
    listener: 
      simple: 
        acknowledge-mode: manual

@Component
@RabbitListener(queues = "order_queue")
public class OrderMQListener {
    /**
     * 处理器，适配器，加上@RabbitHandler注解
     * 加上Channel这个参数
     */
    @RabbitHandler
    public void messageHandler(String body, Message message, Channel channel) throws IOException {
        long tag = message.getMessageProperties().getDeliveryTag();
        System.out.println("message:"+message.toString());
        System.out.println("==============");
        System.out.println("消息标识tag:"+tag);
        System.out.println("消息体body:"+body);
        //第一个参数是该消息的index，第二个是是否批量操作
        channel.basicAck(tag,false);
        //第一个参数是index，第二个是否批量，第三个是失败后是否重新返回给生产者重新投递
        //channel.basicNack(msgTag,false,true);
    }
}