它是一种通过广播方式发送消息的路由器,所有和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学习笔记