@Bean
public ConnectionFactory connectionFactory() {
    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
    cachingConnectionFactory.setVirtualHost("/eugene");
    cachingConnectionFactory.setUsername("guest");
    cachingConnectionFactory.setPassword("guest");
    cachingConnectionFactory.setHost("192.168.111.145");
    cachingConnectionFactory.setPublisherConfirms(true);
    return cachingConnectionFactory;
}

rabbitTemplate.convertAndSend("topic名字", "routingKey", "传递的消息");
// 通过这行代码，rabbitTemplate就能知道将消息发送到哪个交换机上，然后交换机就能根据自己的特性，将routingKey将消息发送给符合条件的队列

@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
    // 将消息发送给RabbitMQ的回调
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        // 第一个参数: 是生产消息是传入的CorrelationData对象，里面维护了一个id，
        // 可以自定义取值来标识某些业务
        // 第二个参数: 判断消息有没有发成功
        // 第三个参数: 发生异常的原因，cause为异常的原因
        System.out.println("ack: " + ack);
        System.out.println("cause: " + cause);
        System.out.println("correlationData: " + correlationData);
    });

    // 允许失败回调
    rabbitTemplate.setMandatory(true);
    // 当消息服务器将消息发送给队列中时的回调
    rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
        // 第一个参数: message -> 消息主体
        // 第二个参数: 发送失败错误码
        // 第三个参数: 发送失败错误信息
        // 第四个参数: 发送的交换机名字
        // 第五个参数: 发送消息的routingKey
        System.out.println(message);
        System.out.println(replyCode);
        System.out.println(replyText);
        System.out.println(exchange);
        System.out.println(routingKey);
    });

    // 自己实现了一个消息转换器
    rabbitTemplate.setMessageConverter(new MessageConverter() {
        @Override
        public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
            // 指定发送的消息类型为test/plain
            messageProperties.setContentType("text/plain");
            // 指定消息的编码格式
            messageProperties.setContentEncoding("UTF-8");
            Message message = new Message(JSON.toJSONBytes(object), messageProperties);
            return message;
        }

        @Override
        public Object fromMessage(Message message) throws MessageConversionException {
            return message;
        }
    });

    return rabbitTemplate;
}

@Bean
public Queue preFetchQueue() {
    return QueueBuilder.durable(Constants.PRE_FETCH_QUEUE_NAME).build();
}

(TopicExchange) ExchangeBuilder.topicExchange("topicExchangeName").withArgument("参数key", "value").build();

BindingBuilder.bind(绑定的队列).to(交换机).with(绑定的key);
// 大致的语法就是将队列绑定到交换机中并监听具体的key

/**
 * messageErrorHandler为一个实现了RabbitListenerErrorHandler接口的消息发生异常的处理器的名字(是一个bean)
 * simpleRabbitListenerContainerFactory是一个实现了RabbitListenerContainerFactory接口的消费者容器工厂
 * (它是一个bean，并且内部设置了消息确认模式为手动)
**/
@RabbitListener(
    queues = Constants.ORDER_QUEUE_NAME,
    errorHandler = "messageErrorHandler",
    containerFactory = "simpleRabbitListenerContainerFactory"
)	

@Component
public class MessageErrorHandler implements RabbitListenerErrorHandler {

    private static final Logger logger = LoggerFactory.getLogger(MessageErrorHandler.class);

    @Override
    public Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message, ListenerExecutionFailedException exception) throws Exception {
        logger.warn("处理消息异常, 异常消息为: {}, 异常信息为: ", amqpMessage, exception);
        logger.info("可以在此处对消息进行持久化存入db，并使用job定时去消费");

        return null;
    }
}

@Bean
public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return factory;
}

// SimpleMessageConverter.java
public Object fromMessage(Message message) throws MessageConversionException {
    Object content = null;
    // 拿到消息的MessageProperties对象
    MessageProperties properties = message.getMessageProperties();
    if (properties != null) {
        String contentType = properties.getContentType();
        // 判断消息的ContentType类型是否为text打头，eg: text/html, text/xml等等
        if (contentType != null && contentType.startsWith("text")) {
            // 拿到消息内部的编码格式，我们可以指定格式。eg: UTF-8, GBK等等
            String encoding = properties.getContentEncoding();
            if (encoding == null) {
                // 默认为utf-8
                encoding = this.defaultCharset;
            }
            try {
                // 最终使用编码格式进行编码
                content = new String(message.getBody(), encoding);
            }
            catch (UnsupportedEncodingException e) {
                throw new MessageConversionException(
                    "failed to convert text-based Message content", e);
            }
        }
        else if (contentType != null &&
                 contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
            try {
                content = SerializationUtils.deserialize(
                    createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
            }
            catch (IOException | IllegalArgumentException | IllegalStateException e) {
                throw new MessageConversionException(
                    "failed to convert serialized Message content", e);
            }
        }
    }
    if (content == null) {
        content = message.getBody();
    }
    return content;
}

@Bean
public TopicExchange topicExchange() {
    Map<String, Object> map = new HashMap<>();
    map.put("alternate-exchange", "defaultExchange");
    return (TopicExchange) ExchangeBuilder.topicExchange(Constants.TOPIC_EXCHANGE).withArguments(map).build();
}

@Bean(Constants.ORDER_QUEUE_NAME)
public Queue queue() {
    return new Queue(Constants.ORDER_QUEUE_NAME);
}

@Bean
public Binding binding() {
    // 将队列和routing key绑定至exchange中, RoutingKey为: order.#
    return BindingBuilder.bind(queue()).to(topicExchange()).with(Constants.ORDER_SERVICE_MATCH_PREFIX_ROUTING_KEY);
}

@Bean
public FanoutExchange defaultExchange() {
    return (FanoutExchange) ExchangeBuilder.fanoutExchange(Constants.DEFAULT_EXCHANGE).build();
}

@Bean
public Binding defaultExchangeBinding() {
    return BindingBuilder.bind(queue()).to(defaultExchange());
}

// routingKey为123order.create
rabbitTemplate.convertAndSend(Constants.TOPIC_EXCHANGE, 123 + Constants.ORDER_CREATE_ROUTING_KEY, JSON.toJSONString(map), correlationData);

reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'alternate-exchange' for exchange 'topicExchange' in vhost '/eugene': received the value 'defaultExchange' of type 'longstr' but current is none, class-id=40, method-id=10