一、介绍
RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至不知道这些消息传递到了哪些队列中。
相反，生产者只能将消息发送到交换机，交换机工作的内容非常简单，一方面他接受来自生产者的消息，另一方面他将他们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们放到许多队列还是说应该丢弃他们。这就由交换机来决定。
二、类型
1、类型
总共有以下类型：直接（direct）[路由],主题（topic），标题（headers）,扇出（fanout）[发布订阅]，
默认类型[无名类型] 通过("")进行标识
channel.basicPublish("",TASK_QUEUE_NAME,null,message.getBytes());第一个参数是交换机名称，空字符串表示默认或无名的交换机；消息能路由发送到队列中，其实是由routingKey(bindingKey)绑定key指定的，如果它存在的话。
2、 临时队列
每当我们连接到Rabbit时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称。其次一旦我们断开了消费者连接，队列将被自动删除。
创建临时队列的方式如下
String queueName = channel.queueDeclare().getQueue();3、绑定（bingings）
binding其实时exchange和queue之间的桥梁，他告诉我们exchange和哪个队列进行了绑定关系
4、fanout
他是将接收到的所有消息广播到他知道的所有队列中
消费者，另一个复制即可
public class ReceiveLogs01 {
//交换机的名称
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtil.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
//声明一个队列 临时队列
/**
* 队列的名称是随机的
* 当消费者断开与队列的连接的时候，队列就自动删除
*/
String queueName = channel.queueDeclare().getQueue();
/**
* 绑定队列与交换机
*/
channel.queueBind(queueName,EXCHANGE_NAME,"");
System.out.println("等待接收消息，把接收到的消息打印在屏幕上......");
//接收消息
DeliverCallback deliverCallback = (consumerTag,message) ->{
System.out.println("01控制台打印接收到的消息:"+new String(message.getBody()));
};
CancelCallback cancelCallback = (consumerTag) ->{};
channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
}
}生产者
public class EmitLog {
//交换机的名称
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
System.out.println("生产者发出消息："+message);
}
}
}结果
5、direct
消息只去到他绑定的routingKey队列中，支持多重绑定，当exchange的绑定类型是direct，但是他绑定的多个队列的key如果都相同，在这种情况下虽然绑定类型是direct但是他表现的就和fanout有点类似了。
生产者
public class DirectLogs {
//交换机的名称
public static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtil.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
channel.basicPublish(EXCHANGE_NAME,"warning",null,message.getBytes());
System.out.println("生产者发出消息："+message);
}
}
}消费者1
public class ReceiveLogs01 {
//交换机名称
public static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtil.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
//声明一个队列
channel.queueDeclare("console",false,false,false,null);
//绑定队列与交换机
channel.queueBind("console",EXCHANGE_NAME,"info");
channel.queueBind("console",EXCHANGE_NAME,"warning");
DeliverCallback deliverCallback = (consumerTag,message) ->{
System.out.println("direct01控制台打印接收到的消息:"+new String(message.getBody()));
};
CancelCallback cancelCallback = (consumerTag) ->{};
channel.basicConsume("console",true,deliverCallback,cancelCallback);
}
}
消费者2
public class ReceiveLogs02 {
//交换机名称
public static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtil.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
//声明一个队列
channel.queueDeclare("disk",false,false,false,null);
//绑定队列与交换机
channel.queueBind("disk",EXCHANGE_NAME,"error");
DeliverCallback deliverCallback = (consumerTag,message) ->{
System.out.println("direct02控制台打印接收到的消息:"+new String(message.getBody()));
};
CancelCallback cancelCallback = (consumerTag) ->{};
channel.basicConsume("disk",true,deliverCallback,cancelCallback);
}
}
