[Java知识库]rabbitmq java/springboot客户端详解——生产者与消费者区别 、channel.exchangeDeclare与channel.queueDeclare区别、交换机和队列如何绑定

这是我的连接工具类：

public class RabbitMqUtils {
    //得到一个连接的 channel
    public static Channel getChannel() throws Exception {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.195.130");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }
}

交换机和队列的声明、绑定在生产者和消费者端都可以。声明是不区分哪一端的，连接的客户端也不区分生产者还是消费者，生产者和消费者的区别只是Channel调用的是basicConsume()还是basicPublish()的区别。 完全可以消费者消费完，然后Channel调用basicPublish()变成生产者进行发送消息。

最好是生产者端声明交换机，消费者端再次声明交换机、声明队列、并进行绑定。队列和交换机需要都存在才能绑定。

其实不需要手动声明交换机和队列都可以，但是Broker中必须要有要使用的交换机和队列。使用交换机和队列的原则是：rabbitmq服务端（Broker）含有交换机和队列，先声明，后使用

只不过为了不影响使用，编写代码时也要声明（declare）一下，再次声明一个交换机/队列不会影响原来存在的。再次声明在rabbitmq服务端（Broker）有就使用，没有就声明一个。

生产者：声明交换机

		Channel channel = RabbitMqUtils.getChannel();
		//没有就声明一个exchange，名称为“AA”，是topic类型
		//有就绑定名称为“AA”的交换机。
        channel.exchangeDeclare("AA", "topic");
        channel.basicPublish("AA", bindingKey, null, "生产者发送消息了");

消费者：
声明交换机和队列，并进行绑定

		Channel channel = RabbitMqUtils.getChannel();
		//没有就声明一个exchange，名称为“AA”，是topic类型
        channel.exchangeDeclare("AA", "topic");
        
        String queueName = "Q1";
        //声明 Q1 队列
        channel.queueDeclare(queueName, false, false, false, null);
        //让交换机“AA” 绑定 队列“Q1”
        channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit");
		//消费者绑定后进行消费，消费者代码就不写了
		

上面只是在普通java工程中演示。在springboot工程中，我们一般把交换机和队列都在配置类中声明和绑定，这样就能和业务代码解耦。

比如如图下一个消息队列架构：

我们可以创建一个这样的配置类，把对应的交换机、队列、已经绑定关系在里面配置好：

package com.atguigu.rabbitmq.bootrabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author CZF
 * @create 2022-09-17 11:33
 */
@Configuration
public class TtlQueueConfig {
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String DEAD_LETTER_QUEUE = "QD"; // 声明 xExchange

    // 声明 xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    } 

    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //声明队列 A ttl 为 10s 并绑定到对应的死信交换
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //声明队列的 TTL
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
    }

    // 声明队列 A 绑定 X 交换机
    @Bean
    public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> args = new HashMap<>(3);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //声明队列的 TTL
        args.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
    }

    //声明队列 B 绑定 X 交换机
    @Bean
    public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
    }

    //声明死信队列 QD
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE);
    }

    //声明死信队列 QD 绑定关系
    @Bean
    public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
                                        @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }
}

然后使用Spring的RabbitTemplate，生产者就可以给消息队列发送消息

rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: "+message);
rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: "+message);

而消费者是通过监听器监听队列的消息，因此要写一个监听器

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;

/**
 * @author CZF
 * @create 2022-09-17 12:50
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {
    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("当前时间：{},收到死信队列信息{}", new Date().toString(), msg);
    }
}