它是一种通过广播方式发送消息的路由器,所有和exchange建立的绑定关系的队列都会接收到消息

不处理路由键,只需要简单的将队列绑定到交换机上

fanout交换机转发消息是最快的,它不需要做路由键的匹配

2. Exchange

Exchange在我们的工作模型中首次出现,因此需要详细介绍下。

Exchange:接收消息,并根据路由键转发消息到所绑定的队列。

Exchange分为4种类型:

Direct:完全根据key进行投递的,例如,绑定时设置了routing key为”abc”,那么客户端提交的消息,只有设置了key为”abc”的才会投递到队列。
Topic:对key进行模式匹配后进行投递,符号”#”匹配一个或多个词,符号”*”匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”,”abc.*”只匹配”abc.def”。
Fanout:不需要key,它采取广播模式,一个消息进来时,投递到与该交换机绑定的所有队列。

交换机属性:

Durable:是否需要持久化,true为持久化

AutoDelete:当最后一个绑定到Exchange上的队列删除后,自动删除该Exchange

Internal:当前Exchange是否用于Rabbitmq的内部使用,默认是false(用Erlang语言进行内部功能扩展会使用到)

Arguments:扩展参数,用于扩展AMQP协议自定义使用

今天我们的实例采用fanout类型的exchange。

3.Queue-消息队列

消息队列,实际存储消息数据
Durability:是否持久化,Durable:true,Transient:false
AutoDelete:true代表当最后一个监听被移除之后,该Queue会自动被删除。

通过RabbitMQ 接口可以自己去生成临时队列,队列名字也由RabbitMQ自动生成。通过

可以声明一个非持久的、通道独占的、自动删除的队列,getQueue()方法可以获取随机队列名字。这个名字用来在队列和exchange之间建立binding关系的时候使用:

4.Message-消息

服务器和应用程序之间传递的数据
本质上就是一段数据,由Properties和Payload(Body)组成
常用属性:
delivery mode:投递模式,2表示持久化投递
headers:自定义属性
content_type:消息内容格式
content_encoding:编码格式
priority:优先级

reply_to:一般用于RPC过程,消息处理返回结果队列;
correlation_id:用于关联RPC的请求与应答;

expiration:过期时间
message_id:消息id

5.创建生产者

package com.dwz.rabbitmq.exchange.fanout;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.dwz.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "test_fanout_exchange";
        String msg = "hello rabbitmq fanout message successs!--";
        for(int i = 0; i < 50; i++) {
            channel.basicPublish(exchangeName, "", null, (msg + i).getBytes());
        channel.close();
        connection.close();

6.创建消费者1

package com.dwz.rabbitmq.exchange.fanout;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.dwz.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        //交换机名称
        String exchangeName = "test_fanout_exchange";
        //交换机类型
        String exchangeType = "fanout";
        String queueName = "test_fanout_queue_1";
        //路由键
        String routingKey = "";
        //交换机声明
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
         * durable:持久化,服务器重启,数据仍然存在
         * exclusive:独占一个channel,用于顺序消费,类似于加了一把锁
         * autoDelete:如果一个队列脱离了exchange会自动删除
        //队列声明
        channel.queueDeclare(queueName, false, false, false, null);
        //队列绑定交换机
        channel.queueBind(queueName, exchangeName, routingKey);
        //定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("rec1--message:" + msg);
        //设置消费者
        channel.basicConsume(queueName, true, consumer);

7.创建消费者2

package com.dwz.rabbitmq.exchange.fanout;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.dwz.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "test_fanout_exchange";
        String exchangeType = "fanout";
        String queueName = "test_fanout_queue_2";
        String routingKey = "";
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
         * durable:持久化,服务器重启,数据仍然存在
         * exclusive:独占一个channel,用于顺序消费,类似于加了一把锁
         * autoDelete:如果一个队列脱离了exchange会自动删除
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("rec2--message:" + msg);
        channel.basicConsume(queueName, true, consumer);

8.运行代码

先启动两个消费者,再启动生产者,运行结果:

两个消费者都拿到exchange(交换机)的全部信息

注:由于exchange只是做一个消息转发,本身是不具备存储数据的能力,如果先启动生产者,信息是到达不了队列的,队列有存储数据的功能

 参考文章:

RabbitMQ学习笔记