MQ全称为Message Queue,即消息队列, RabbitMQ是由erlang语言开发,基于AMQP(Advanced MessageQueue 高级消息队列协议)协议实现的消息队列,它是一种应用程序之间的通信方法,消息队列在分布式系统开发中应用非常广泛。RabbitMQ官方地址:http://www.rabbitmq.com/,
优点:
- Spring Boot默认已集成RabbitMQ,开发快捷便利
- 使用简单,功能强大。
- 基于AMQP协议。
- 高并发性能好,基于Erlang语言。
RabbitMQ属性的简单说明:
- Broker:消息队列服务进程,此进程包括两个部分:Exchange和Queue。
- Exchange:消息队列交换机,按一定的规则将消息路由转发到某个队列,对消息进行过虑。
- Queue:消息队列,存储消息的队列,消息到达队列并转发给指定的消费方。
- Producer:消息生产者,即生产方客户端,生产方客户端将消息发送到MQ。
- Consumer:消息消费者,即消费方客户端,接收MQ转发的消息。
消息发布接收流程:
-----发送消息----- | ----接收消息----- |
---|---|
1,生产者和Broker建立TCP连接 | 1,消费者和Broker建立TCP连接 |
2,生产者和Broker建立通道 | 2,消费者和Broker建立通道 |
3,生产者通过通道消息发送给Broker,由Exchange将消息进行转发 | 3, 消费者监听指定的Queue(队列) |
4, Exchange将消息转发到指定的Queue(队列) | 4,当有消息到达Queue时Broker默认将消息推送给消费者。 |
5,消费者接收到消息。 |
引入依赖:
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp‐client</artifactId>
<version>自定义</version><!‐‐必须与自己SpringBoot版本信息相匹配‐‐>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring‐boot‐starter‐logging</artifactId>
</dependency>
创建生产者:
public class Producer {
private static final String QUEUE_Name = "newRabbitMQ";//队列名称
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = null;//新建连接,设置为null,备用
Channel channel = null;//新建通道,设置为null,备用
try
{
ConnectionFactory factory = new ConnectionFactory();//定义连接工厂
factory.setHost("");//RabbitMQ安装服务器的IP地址,如果是本机则为127.0.0.1
factory.setPort(5672);//java程序连接时的默认端口,浏览器为15672
factory.setUsername("guest");//RabbitMQ默认用户名
factory.setPassword("guest");//RabbitMQ默认密码
factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”,虚拟机相当于一个独立的mq服务
器
//创建与RabbitMQ服务的TCP连接
connection = factory.newConnection();
//创建与Exchange的通道,每个连接可以创建多个通道,每个通道代表一个会话任务
channel = connection.createChannel();
/*声明队列,如果Rabbit中没有此队列将自动创建*/
channel.queueDeclare(QUEUE_Name/*队列名称*/,
true/*是否持久化*/,
false/*队列是否独占此连接*/,
false/*队列不再使用时是否自动删除此队列*/,
null/*队列参数*/);
String message = "hello,RabbitMQ"+System.currentTimeMillis();//编辑要发送的信息,并输出系统时间
/**
* 这里没有指定交换机,消息将发送给默认交换机,每个队列也会绑定那个默认的交换机,但不能显示绑定或解除绑定
*默认的交换机,routingKey等于队列名称
*/
channel.basicPublish/*消息发布方法 */(""/*Exchange的名称,如果没有指定,则使用Default Exchange*/,
QUEUE/*routingKey,消息的路由Key,是用于Exchange(交换机)将消息转发到指定的消息队列*/,
null/*消息包含的属性*/,
message.getBytes())/*消息体*/;
System.out.println("Send Message is:'" + message + "'");//输出发送的信息
}catch(Exception ex){
ex.printStackTrace();//输出异常信息
}
finally{
if(channel != null){
channel.close();//关闭释放通道
}
if(connection != null){
connection.close();//关闭释放连接
}
}
}
}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
编辑完毕,后选择运行,向RabbitMQ发送第一条信息,然后进行后续代码操作
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
创建消费者:
public class Consumer {
private static final String QUEUE_Name = "newRabbitMQ";
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();//定义连接工厂
factory.setHost(""); //RabbitMQ安装服务器的IP地址,如果是本机则为127.0.0.1
factory.setPort(5672);//java程序连接时的默认端口,浏览器为15672
Connection connection = factory.newConnection();//新建连接
Channel channel = connection.createChannel();//新建通道
channel.queueDeclare(QUEUE_Name, true, false, false, null); //声明队列,参数信息同上
DefaultConsumer consumer = new DefaultConsumer(channel) //定义消费方法{
@Override
public void handleDelivery(String consumerTag,//消费者的标签,在channel.basicConsume()去指定
Envelope envelope,//消息包的内容,可从中获取消息id,消息routingkey,交换机,消息和重传标志,如下所示(收到消息失败后是否需要重新发送)
AMQP.BasicProperties properties,//基础配置信息
byte[] body//队列发送的信息
)throws IOException {
String exchange = envelope.getExchange(); //交换机
String routingKey = envelope.getRoutingKey();//路由key
long deliveryTag = envelope.getDeliveryTag();//消息id
String msg = new String(body,"utf‐8");//消息内容
System.out.println("receive message.." + msg);
}
};
channel.basicConsume(QUEUE_Name,/*队列名称*/
false,/*是否自动回复,设置为true为表示消息接收到自动向mq回复接收到了,mq接收到回复会删除消息,设置
为false则需要手动回复,为了避免消息丢失,建议选择手动回复!*/
consumer/*消费消息的方法,消费者接收到消息后调用此方法*/
);
}
}
+++++++++++++++++++++++++++++++++++++++++++++++++++
编辑完毕,后选择运行,向RabbitMQ接收生产者发送的信息
+++++++++++++++++++++++++++++++++++++++++++++++++++
运行流程:
步骤 | 流程信息 |
---|---|
生产者流程 | 生产者操作流程信息 |
1 | 创建连接 |
2 | 创建通道 |
3 | 声明队列 |
4 | 发送消息 |
消费者流程 | 消费者者操作流程信息 |
1 | 创建连接 |
2 | 创建通道 |
3 | 声明队列 |
4 | 监听队列 |
5 | 接收消息 |
6 | ack回复 |
结言(封装配置信息):
上述代码虽然较为繁琐,但是可为新手朋友构建较为清晰明了的运行流程,你也可对部分代码进行封装,以提高代码的复用性!(以下仅供参考,可根据自身情况进行操作)
public static Connection getConnection() throws Exception {
ConnectionFactory factory = new ConnectionFactory(); //定义连接工厂
factory.setHost("");//RabbitMQ安装服务器的IP地址,如果是本机则为127.0.0.1
factory.setPort(5672);//java程序连接时的默认端口,浏览器为15672
factory.setVirtualHost("/");//rabbitmq默认虚拟机
factory.setUsername("guest");//默认用户名
factory.setPassword("guest");//默认密码
Connection connection = factory.newConnection();// 通过工程获取连接
return connection;//返回连接信息
}
祝君日进斗金,事业有成