@Configuration
public class RabbitConfig {

    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Book implements Serializable {

    private static final long serialVersionUID = -2647663884508887444L;
    private int id;
    private String name;
    private String author;

}

@Component
@Slf4j
public class DirectReceiver {

    /**
     * 简单直连消息接收
     * @param message
     */
    @RabbitListener(queuesToDeclare=@Queue("direct.queue"))
    @RabbitHandler
    public void message(Message message) {
        log.info("message result = {}",message);
    }


}

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringBootRabbitmqExampleApplicationTests {
    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    public void contextLoads() {
        Book book = new Book(1,"spring cloud","mtcarpenter");
        amqpTemplate.convertAndSend("direct.queue",book);
    }

}

2020-05-25 15:00:09.444  INFO 1456 --- [ntContainer#0-1] c.m.r.example.direct.DirectReceiver      : message result = (Body:'{"id":1,"name":"spring cloud","author":"mtcarpenter"}' MessageProperties [headers={__TypeId__=com.mtcarpenter.rabbitmq.example.Book}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=direct.queue, deliveryTag=1, consumerTag=amq.ctag-5LNFFWiSMM7C338m1SSRvg, consumerQueue=direct.queue])

/**
 * 消息接收者 - consumer
 * 
 * @RabbitListener - 可以注解类和方法。
 *  注解类，当表当前类的对象是一个rabbit listener。
 *      监听逻辑明确，可以由更好的方法定义规范。
 *      必须配合@RabbitHandler才能实现rabbit消息消费能力，一个类可以有多个方法，但是仅有一个方法注解@RabbitHandler。
 *  注解方法，代表当前方法是一个rabbit listener处理逻辑。
 *      方便开发，一个类中可以定义若干个listener逻辑。
 *      方法定义规范可能不合理。如：一个方法的处理逻辑太多，造成方法的bad smell。
 * 
 * @RabbitListener -  代表当前类型是一个rabbitmq的监听器。
 *      bindings:绑定队列
 * @QueueBinding  - @RabbitListener.bindings属性的类型。绑定一个队列。
 *      value:绑定队列， Queue类型。
 *      exchange:配置交换器， Exchange类型。
 *      key:路由键，字符串类型。
 * 
 * @Queue - 队列。
 *      value:队列名称
 *      autoDelete:是否是一个临时队列。
 *          true ：当所有的consumer关闭后，自动删除queue。
 *          false：当任意一个consumer启动并创建queue后，如果queue中有消息未消费，无论是否有consumer继续执行，都保存queue。
 * 
 * @Exchange - 交换器
 *      value:为交换器起个名称
 *      type:指定具体的交换器类型
 */
@Component
@Slf4j
public class TopicReceiver {


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "magazineOrder",durable = "true"),
            exchange = @Exchange(value = "order",type = "topic"),
            key = "magazine"
    ))
    @RabbitHandler
    public void magazineMessage(Message message) {
        log.info("magazine result = {}",message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "historyOrder",durable = "true"),
            exchange = @Exchange(value = "order",type = "topic"),
            key = "history"
    ))
    @RabbitHandler
    public void historyMessage(Message message) {
        log.info("history result = {}",message);
    }
}

@Test
public void topicReceiver(){
    Book magazine = new Book(1,"杂志图书","mtcarpenter1");
    amqpTemplate.convertAndSend("order","magazine",magazine);
    Book history = new Book(2,"历史图书","mtcarpenter2");
    amqpTemplate.convertAndSend("order","history",history);
}

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "encyclopediaOrder",durable = "true"),
            exchange = @Exchange(value = "order",type = "topic",
                    ignoreDeclarationExceptions = "true"),
            key = "encyclopedia"
    ))
    @RabbitHandler
    public void onOrderMessage(@Payload Book book,
                               Channel channel,
                               @Headers Map<String, Object> headers) throws Exception {
       log.info("book id = {}" + book.getId());
        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        //手工ACK
        channel.basicAck(deliveryTag, false);
    }

@Component
@Slf4j
public class TopicSender implements RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 发送消息方法调用: 构建Message消息
     * @param book
     * @throws Exception
     */
    public void send(Book book) throws Exception {
        //设置消息退回后的回调处理机制
        this.rabbitTemplate.setReturnCallback(this);
        // CorrelationData是一个当发送原始消息时，由客户机提供的对象。
        // ack是一个boolean值，当ack（确认）的时候，值为true；当nack（不确认）的时候，值为false
        // 经测试发现，只要 rabbitTemplate.convertAndSend() 能正确找到exchange，无论是否能将消息路由到正确的queue,ack值都为true
        // 只有当rabbitTemplate.convertAndSend()无法找到exchange时，ack 值才为false
        // cause是附加原因，比如说当nack的时候附加的原因

        this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败 cause = {} ,correlationData={} ", cause, correlationData.toString());
            } else {
                log.info("消息发送成功 ");
            }
        });
        this.rabbitTemplate.convertAndSend("order", "encyclopedia", book);
    }


    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("return message={} , exchange = {} , routingKey = {},replyCode={}, replyText{}",
                message, exchange, routingKey, replyCode, replyText);

    }
}

@Test
public void topicSender() throws Exception {
    topicSender.send(new Book(3,"百科图书","mtcarpenter"));
}