fanout 扇出交换机

Fanout 这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的 所有队列中。系统中默认有些 exchange 类

代码

生产者

import com.gao.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * author:      高亮
 * time:        2022/7/6
 * description: 扇出交换机,生产者
 */
@Slf4j
public class ExchangeFanout_Producer {
    public static final String EXCHANGE_FANOUT = "fanout_exchange";
    public static final int MESSAGE_COUNT = 30;

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();
        /**
         * 声明交换机
         * params1: 交换机名称
         * params2：交换机类型
         */
        channel.exchangeDeclare(EXCHANGE_FANOUT, "fanout");
        /**
         *
         * 声明队列 临时
         * 生成一个临时队列、队列的名称是随机的
         * 当消费者断开与队列的连接的时候队列就自动删除
         */
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue, EXCHANGE_FANOUT, "");
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "张三"+i;
            channel.basicPublish(EXCHANGE_FANOUT, "", null, message.getBytes(StandardCharsets.UTF_8));
        }
        channel.close();

    }
}

消费者

import com.gao.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * author:      高亮
 * time:        2022/7/6
 * description: 扇出交换机,消费者
 */
@Slf4j
public class ExchangeFanout_Consumer {
    public static final String EXCHANGE_FANOUT = "fanout_exchange";

    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_FANOUT, "fanout");
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue,EXCHANGE_FANOUT,"");
        DeliverCallback deliverCallback = (consumerTag,message)->{
            log.info("接收到的消息: {}", new String(message.getBody(), StandardCharsets.UTF_8));
        };
        CancelCallback cancelCallback = (consumerTag)->{
            log.info("取消的消息: {}", consumerTag);
        };

        channel.basicConsume(queue, true, deliverCallback, cancelCallback);
        log.info("等待接收消息");
    }
}