RabbitMQ 一文读懂
Posted 在奋斗的大道
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ 一文读懂相关的知识，希望对你有一定的参考价值。
目录
第一步：创建Maven项目，添加RabbitMQ 消息队列依赖。
1、RabbitMQ 介绍
MQ 全称为 Message Queue ，即消息队列， RabbitMQ是由 erlang 语言开发，基于 AMQP （ Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。RabbitMQ 官方地址： http://www.rabbitmq.com/应用场景
1 、任务异步处理。 将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。 2 、应用程序解耦合 MQ相当于一个中介，生产方通过 MQ 与消费方交互，它将应用程序进行解耦合。其他消息队列
ActiveMQ ， RabbitMQ ， ZeroMQ ， Kafka ， MetaMQ ， RocketMQ 、 Redis 。选择RabbitMQ 原因
1 、使得简单，功能强大。 2 、基于 AMQP 协议。 3 、社区活跃，文档完善。 4 、高并发性能好，这主要得益于 Erlang 语言。 5 、 Spring Boot 默认已集成 RabbitMQ2、AMQP消息队列其他相关知识
什么是AMQP？
AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。
什么是JMS？
JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。
3、RabbitMQ 快速入门
RabbitMQ的工作原理
组成部分说明如下：- Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
- Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
- Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
- Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。
- Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。
RabbitMQ 消息发送和接受流程梳理
RabbitMQ 消息发送
1 、生产者和 Broker 建立 TCP 连接。 2 、生产者和 Broker 建立通道。 3 、生产者通过通道消息发送给 Broker ，由 Exchange 将消息进行转发。 4 、 Exchange 将消息转发到指定的 Queue （队列）
RabbitMQ 消息接受
1 、消费者和 Broker 建立 TCP 连接 2 、消费者和 Broker 建立通道 3 、消费者监听指定的 Queue （队列） 4 、当有消息到达 Queue 时 Broker 默认将消息推送给消费者。 5 、消费者接收到消息。
RabbitMQ 安装
RabbitMQ 之Hello World
第一步：创建Maven项目，添加RabbitMQ 消息队列依赖。
<!--集成RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>第二步：创建RabbitMQ 消息生产者
package com.zzg.rabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ProduceMS
//队列名称
private static final String QUEUE = "helloworld";
public static void main(String[] args) throws IOException, TimeoutException
Connection connection = null;
Channel channel = null;
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.43.10");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
//创建与RabbitMQ服务的TCP连接
connection = factory.newConnection();
//创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
channel = connection.createChannel();
/*** 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* */
channel.queueDeclare(QUEUE, true, false, false, null);
String message = "Hello World RabbitMQ" + System.currentTimeMillis();
/*** 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
* */
/*** 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显 示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
* */
channel.basicPublish("", QUEUE, null, message.getBytes());
System.out.println("Send Message is:'" + message + "'");
if (channel != null)
channel.close();
if (connection != null)
connection.close();
第三步: 创建RabbitMQ 消息消费者
package com.zzg.rabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class ConsumerMS
private static final String QUEUE = "helloworld";
public static void main(String[] args) throws IOException, TimeoutException
ConnectionFactory factory = new ConnectionFactory();
//设置RabbitMQ所在服务器的ip和端口
factory.setHost("192.168.43.10");
factory.setPort(5672);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE, true, false, false, null);
//定义消费方法
DefaultConsumer consumer = new DefaultConsumer(channel)
/*** 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志 (收到消息失败后是否需要重新发送)
* @param properties
* @param body
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
//消息id
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String msg = new String(body, "utf-8");
System.out.println("receive message.." + msg);
;
/*** 监听队列String queue, boolean autoAck,Consumer callback
* 参数明细
* 1、队列名称
* 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复
* 3、消费消息的方法，消费者接收到消息后调用此方法
*/
channel.basicConsume(QUEUE, true, consumer);
4、RabbitMQ 支持工作模式
RabbitMQ 有以下几种工作模式 ：- Work queues
- Publish/Subscribe
- Routing
- Topics
- Header
- RPC
Work queues模式
多个消费端消费同一个队列中的消息，队列采用轮询的方式将消息是平均发送给消费者；
缺少图片
特点：
1、一条消息只会被一个消费端接收；
2、队列采用轮询的方式将消息是平均发送给消费者的；
3、消费者在处理完某条消息后，才会收到下一条消息
Work Queues模式之生产者
package com.zzg.rabbitmq.work.queues;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
生产者核心步骤
1、声明队列
2、创建连接
3、创建通道
4、通道声明队列
5、制定消息
6、发送消息，使用默认交换机
*/
public class WorkQueueProduceMS
//声明队列
private static final String QUEUE = "queue";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");//mq服务ip地址
connectionFactory.setPort(5672);//mq client连接端口
connectionFactory.setUsername("guest");//mq登录用户名
connectionFactory.setPassword("guest");//mq登录密码
connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
//创建与RabbitMQ服务的TCP连接
connection = connectionFactory.newConnection();
//创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
channel = connection.createChannel();
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE, true, false, false, null);//通道绑定邮件队列
for (int i = 0; i < 10; i++)
String message = new String("mq 发送消息。。。");
/**
* 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
* 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
*/
//String exchange, String routingKey, BasicProperties props, byte[] body
channel.basicPublish("", QUEUE, null, message.getBytes("utf-8"));
System.out.println("mq消息发送成功！");
catch (Exception e)
e.printStackTrace();
finally
try
channel.close();
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
try
connection.close();
catch (IOException e)
e.printStackTrace();
Work Queues模式之消费者
package com.zzg.rabbitmq.work.queues;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/*
消费者核心步骤
1、声明队列
2、创建连接
3、创建通道
4、通道声明队列
5、重写消息消费方法
6、执行消息方法
*/
public class WorkQueueConsumerMS
private static final String QUEUE = "queue";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");
connectionFactory.setPort(5672);
connection = connectionFactory.newConnection();
channel = connection.createChannel();
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE, true, false, false, null);//通道绑定邮件队列
//String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
DefaultConsumer consumer = new DefaultConsumer(channel)
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @param properties
* @param body
* @throws IOException
* String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
envelope.getDeliveryTag();
String msg = new String(body, "utf-8");
System.out.println("mq收到的消息是：" + msg);
;
System.out.println("消费者启动成功！");
channel.basicConsume(QUEUE, true, consumer);
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
生产端启动后，控制台打印信息如下：
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！
mq消息发送成功！RabbitMQ中的已有消息：
缺失图片
消费端启动后，控制台打印信息如下：
消费者启动成功！
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
mq收到的消息是：mq 发送消息。。。
Publish/subscribe 模式
这种模式又称为发布订阅模式，相对于Work queues模式，该模式多了一个交换机，生产端先把消息发送到交换机，再由交换机把消息发送到绑定的队列中，每个绑定的队列都能收到由生产端发送的消息。
缺少图片
特点：
1、每个消费者监听自己的队列；
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收
到消息
应用场景：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法；
publish/subscribe 模式之生产者
package com.zzg.rabbitmq.publish.subscribe;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 生产端核心步骤：
* <p>
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声明交换机
* <p>
* 5、通道声明队列
* <p>
* 6、通过通道使队列绑定到交换机
* <p>
* 7、制定消息
* <p>
* 8、发送消息
*/
public class PublicSubProduceMS
//声明两个队列和一个交换机
//Publish/subscribe发布订阅模式
private static final String QUEUE_EMAIL = "queueEmail";
private static final String QUEUE_SMS = "queueSms";
private static final String EXCHANGE = "messageChange";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");//mq服务ip地址
connectionFactory.setPort(5672);//mq client连接端口
connectionFactory.setUsername("guest");//mq登录用户名
connectionFactory.setPassword("guest");//mq登录密码
connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
//创建与RabbitMQ服务的TCP连接
connection = connectionFactory.newConnection();
//创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
channel = connection.createChannel();
//通道绑定交换机
/**
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
//Publish/subscribe发布订阅模式
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
//交换机和队列绑定
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
//Publish/subscribe发布订阅模式
channel.queueBind(QUEUE_EMAIL, EXCHANGE, "");
channel.queueBind(QUEUE_SMS, EXCHANGE, "");
for (int i = 0; i < 10; i++)
String message = new String("mq 发送消息。。。");
/**
* 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
* 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
*/
//String exchange, String routingKey, BasicProperties props, byte[] body
//Publish/subscribe发布订阅模式
channel.basicPublish(EXCHANGE, "", null, message.getBytes());
System.out.println("mq消息发送成功！");
catch (Exception e)
e.printStackTrace();
finally
try
channel.close();
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
try
connection.close();
catch (IOException e)
e.printStackTrace();
publish/subscribe 模式之消费者（邮件）
package com.zzg.rabbitmq.publish.subscribe;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者核心步骤
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声明交换机
* <p>
* 5、通道声明队列
* <p>
* 6、通过通道使队列绑定到交换机
* <p>
* 7、重写消息消费方法
* <p>
* 8、执行消息方法
*/
public class PublicSubEMailComsumerMS
//Publish/subscribe发布订阅模式
private static final String QUEUE_EMAIL = "queueEmail";
private static final String EXCHANGE = "messageChange";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");
connectionFactory.setPort(5672);
connection = connectionFactory.newConnection();
channel = connection.createChannel();
//通道绑定交换机
/**
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
//Publish/subscribe发布订阅模式
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
//交换机和队列绑定
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
//Publish/subscribe发布订阅模式
channel.queueBind(QUEUE_EMAIL, EXCHANGE, "");
//String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
DefaultConsumer consumer = new DefaultConsumer(channel)
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @param properties
* @param body
* @throws IOException
* String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
envelope.getDeliveryTag();
String msg = new String(body, "utf-8");
System.out.println("mq收到的消息是：" + msg);
;
System.out.println("消费者启动成功！");
channel.basicConsume(QUEUE_EMAIL, true, consumer);
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
publish/subscribe 模式之消费者（短信）
package com.zzg.rabbitmq.publish.subscribe;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者核心步骤
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声明交换机
* <p>
* 5、通道声明队列
* <p>
* 6、通过通道使队列绑定到交换机
* <p>
* 7、重写消息消费方法
* <p>
* 8、执行消息方法
*/
public class PubSubSMSComsumerMS
//Publish/subscribe发布订阅模式
private static final String QUEUE_SMS = "queueSms";
private static final String EXCHANGE = "messageChange";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");
connectionFactory.setPort(5672);
connection = connectionFactory.newConnection();
channel = connection.createChannel();
//通道绑定交换机
/**
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
//Publish/subscribe发布订阅模式
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
//交换机和队列绑定
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
//Publish/subscribe发布订阅模式
channel.queueBind(QUEUE_SMS, EXCHANGE, "");
DefaultConsumer consumer = new DefaultConsumer(channel)
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @param properties
* @param body
* @throws IOException
* String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
envelope.getDeliveryTag();
String msg = new String(body, "utf-8");
System.out.println("mq收到的消息是：" + msg);
;
System.out.println("消费者启动成功！");
channel.basicConsume(QUEUE_SMS, true, consumer);
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
Routing 路由模式
Routing 模式又称路由模式，该种模式除了要绑定交换机外，发消息的时候还要制定routing key，即路由key，队列通过通道绑定交换机的时候，需要指定自己的routing key，这样，生产端发送消息的时候也会指定routing key，通过routing key就可以把相应的消息发送到绑定相应routing key的队列中去。
缺失图片
特点：
1、每个消费者监听自己的队列，并且设置routingkey；
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列；
应用场景：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法；
routing路由模式之生产者
package com.zzg.rabbitmq.routing;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 生产者核心步骤
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声明交换机
* <p>
* 5、通道声明队列
* <p>
* 6、通过通道使队列绑定到交换机并指定该队列的routingkey
* <p>
* 7、制定消息
* <p>
* 8、发送消息并指定routingkey
*/
public class RoutingProduceMS
//声明两个队列和一个交换机
//Routing 路由模式
private static final String QUEUE_EMAIL = "queueEmail";
private static final String QUEUE_SMS = "queueSms";
private static final String EXCHANGE = "routingMessageChange";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");//mq服务ip地址
connectionFactory.setPort(5672);//mq client连接端口
connectionFactory.setUsername("guest");//mq登录用户名
connectionFactory.setPassword("guest");//mq登录密码
connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
//创建与RabbitMQ服务的TCP连接
connection = connectionFactory.newConnection();
//创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
channel = connection.createChannel();
//通道绑定交换机
/**
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
//Routing 路由模式
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
channel.queueDeclare(QUEUE_SMS, true, false, false, null);//通道绑定短信队列
//交换机和队列绑定
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
//Routing 路由模式
channel.queueBind(QUEUE_EMAIL, EXCHANGE, QUEUE_EMAIL);
channel.queueBind(QUEUE_SMS, EXCHANGE, QUEUE_SMS);
//给email队列发消息
for (int i = 0; i < 10; i++)
String message = new String("mq 发送email消息。。。");
/**
* 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
* 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
*/
//String exchange, String routingKey, BasicProperties props, byte[] body
//Routing 路由模式
channel.basicPublish(EXCHANGE, QUEUE_EMAIL, null, message.getBytes());
System.out.println("mq消息发送成功！");
//给sms队列发消息
for (int i = 0; i < 10; i++)
String message = new String("mq 发送sms消息。。。");
/**
* 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
* 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
*/
//String exchange, String routingKey, BasicProperties props, byte[] body
//Routing 路由模式
channel.basicPublish(EXCHANGE, QUEUE_SMS, null, message.getBytes());
System.out.println("mq消息发送成功！");
catch (Exception e)
e.printStackTrace();
finally
try
channel.close();
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
try
connection.close();
catch (IOException e)
e.printStackTrace();
routing路由模式之消费者（邮件）
package com.zzg.rabbitmq.routing;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者核心步骤
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声明交换机
* <p>
* 5、通道声明队列
* <p>
* 6、通过通道使队列绑定到交换机并指定routingkey
* <p>
* 7、重写消息消费方法
* <p>
* 8、执行消息方法
*/
public class RoutingEMailConsumerMS
//Routing 路由模式
private static final String QUEUE_EMAIL = "queueEmail";
private static final String EXCHANGE = "routingMessageChange";
public static void main(String[] args)
Connection connection = null;
Channel channel = null;
try
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.43.10");
connectionFactory.setPort(5672);
connection = connectionFactory.newConnection();
channel = connection.createChannel();
//通道绑定交换机
/**
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
//Routing 路由模式
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
//通道绑定队列
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数
* String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
*
*/
channel.queueDeclare(QUEUE_EMAIL, true, false, false, null);//通道绑定邮件队列
//交换机和队列绑定
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
//Routing 路由模式
channel.queueBind(QUEUE_EMAIL, EXCHANGE, QUEUE_EMAIL);
//String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
DefaultConsumer consumer = new DefaultConsumer(channel)
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @param properties
* @param body
* @throws IOException
* String consumerTag, Envelope envelope, BasicProperties properties, byte[] body
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
//交换机
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
envelope.getDeliveryTag();
String msg = new String(body, "utf-8");
System.out.println("mq收到的消息是：" + msg);
;
System.out.println("消费者启动成功！");
channel.basicConsume(QUEUE_EMAIL, true, consumer);
catch (IOException e)
e.printStackTrace();
catch (TimeoutException e)
e.printStackTrace();
routing路由模式之消费者（短信）
package com.zzg.rabbitmq.routing;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* 消费者核心步骤
* 1、声明队列，声明交换机
* <p>
* 2、创建连接
* <p>
* 3、创建通道
* <p>
* 4、通道声以上是关于RabbitMQ 一文读懂的主要内容，如果未能解决你的问题，请参考以下文章