rabbitmq实战 php,【成都校区】RabbitMQ实战教程

我就是我 2023-01-17 08:28 149阅读 0赞

本帖最后由 小刀葛小伦 于 2019-11-21 18:06 编辑

1.什么是MQ

消息队列(Message Queue，简称MQ)，从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已。

其主要用途：不同进程Process/线程Thread之间通信。

为什么会产生消息队列？有几个原因：

不同进程(process)之间传递消息时，两个进程之间耦合程度过高，改动一个进程，引发必须修改另一个进程，为了隔离这两个进程，在两进程间抽离出一层(一个模块)，所有两进程之间传递的消息，都必须通过消息队列来传递，单独修改某一个进程，不会影响另一个；

不同进程(process)之间传递消息时，为了实现标准化，将消息的格式规范化了，并且，某一个进程接受的消息太多，一下子无法处理完，并且也有先后顺序，必须对收到的消息进行排队，因此诞生了事实上的消息队列；

关于消息队列的详细介绍请参阅：

《Java帝国之消息队列》

《一个故事告诉你什么是消息队列》

《到底什么时候该使用MQ》

MQ框架非常之多，比较流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里开源的RocketMQ。本文主要介绍RabbitMq。

2.RabbitMQ

2.1.RabbitMQ的简介

开发语言：Erlang – 面向并发的编程语言。

2.1.1.AMQP

AMQP是消息队列的一个协议。

2.2.官网

2.3.MQ的其他产品

2.4.学习5种队列

2.5.安装文档

3.搭建RabbitMQ环境

3.1.下载

下载地址：http://www.rabbitmq.com/download.html

3.2.windows下安装

3.2.1.安装Erlang

下载：http://www.erlang.org/download/otp\_win64\_17.3.exe

安装：

安装完成。

3.2.2.安装RabbitMQ

安装完成。

开始菜单里出现如下选项：

启动、停止、重新安装等。

3.2.3.启用管理工具

1、双击

2、进入C:\\Program Files (x86)\\RabbitMQ Server\\rabbitmq\_server-3.4.1\\sbin输入命令：

rabbitmq-plugins enable rabbitmq\_management

这样就启动了管理工具，可以试一下命令：

停止：net stop RabbitMQ

启动：net start RabbitMQ

3、在浏览器中输入地址查看：http://127.0.0.1:15672/

4、使用默认账号登录：guest/ guest

3.3.Linux下安装

3.3.1.安装Erlang

3.3.2.添加yum支持

cd /usr/local/src/

mkdir rabbitmq

cd rabbitmq

wget http://packages.erlang-solutions ... ns-1.0-1.noarch.rpm

rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

rpm --import http://packages.erlang-solutions.com/rpm/erlang\_solutions.asc

使用yum安装：

sudo yum install erlang

3.3.3.安装RabbitMQ

上传rabbitmq-server-3.4.1-1.noarch.rpm文件到/usr/local/src/rabbitmq/

安装：

rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

3.3.4.启动、停止

service rabbitmq-server start

service rabbitmq-server stop

service rabbitmq-server restart

3.3.5.设置开机启动

chkconfig rabbitmq-server on

3.3.6.设置配置文件

cd /etc/rabbitmq

cp /usr/share/doc/rabbitmq-server-3.4.1/rabbitmq.config.example /etc/rabbitmq/

mv rabbitmq.config.example rabbitmq.config

3.3.7.开启用户远程访问

vi /etc/rabbitmq/rabbitmq.config

注意要去掉后面的逗号。

3.3.8.开启web界面管理工具

rabbitmq-plugins enable rabbitmq\_management

service rabbitmq-server restart

3.3.9.防火墙开放15672端口

/sbin/iptables -I INPUT -p tcp --dport 15672 -j ACCEPT

/etc/rc.d/init.d/iptables save

3.4.安装的注意事项

1、推荐使用默认的安装路径

2、系统用户名必须是英文

Win10改名字非常麻烦，具体方法百度

3、计算机名必须是英文

4、系统的用户必须是管理员

如果安装失败应该如何解决：

1、重装系统 – 不推荐

2、将RabbitMQ安装到linux虚拟机中

a)推荐

3、使用别人安装好的RabbitMQ服务

a)只要给你开通一个账户即可。

b)使用公用的RabbitMQ服务，在192.168.50.22

c)推荐

常见错误：

3.5.安装完成后操作

1、系统服务中有RabbitMQ服务，停止、启动、重启

2、打开命令行工具

如果找不到命令行工具,直接cd到相应目录：

输入命令rabbitmq-plugins enable rabbitmq\_management启用管理插件

查看管理页面

通过默认账户 guest/guest 登录

如果能够登录，说明安装成功。

4.添加用户

4.1.添加admin用户

4.2.用户角色

1、超级管理员(administrator)

可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

2、监控者(monitoring)

可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

3、策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、普通管理者(management)

仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

5、其他

无法登陆管理控制台，通常就是普通的生产者和消费者。

4.3.创建Virtual Hosts

选中Admin用户，设置权限：

看到权限已加：

4.4.管理界面中的功能

5.学习五种队列

5.1.导入my-rabbitmq项目

项目下载地址：

https://download.csdn.net/download/zpcandzhj/10585077

5.2.简单队列

5.2.1.图示

P：消息的生产者

C：消息的消费者

红色：队列

生产者将消息发送到队列，消费者从队列中获取消息。

5.2.2.导入RabbitMQ的客户端依赖

com.rabbitmq

amqp-client

3.4.1

5.2.3.获取MQ的连接

package com.zpc.rabbitmq.util;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.Connection;

public class ConnectionUtil \{

public static Connection getConnection() throws Exception \{

//定义连接工厂

ConnectionFactory factory = new ConnectionFactory();

//设置服务地址

factory.setHost("localhost");

//端口

factory.setPort(5672);

//设置账号信息，用户名、密码、vhost

factory.setVirtualHost("testhost");

factory.setUsername("admin");

factory.setPassword("admin");

// 通过工程获取连接

Connection connection = factory.newConnection();

return connection;

5.2.4.生产者发送消息到队列

package com.zpc.rabbitmq.simple;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

public class Send \{

private final static String QUEUE\_NAME = "q\_test\_01";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

// 从连接中创建通道

Channel channel = connection.createChannel();

// 声明(创建)队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 消息内容

String message = "Hello World!";

channel.basicPublish("", QUEUE\_NAME, null, message.getBytes());

System.out.println(" \[x\] Sent '" + message + "'");

//关闭通道和连接

channel.close();

connection.close();

5.2.5.管理工具中查看消息

点击上面的队列名称，查询具体的队列中的信息：

5.2.6.消费者从队列中获取消息

package com.zpc.rabbitmq.simple;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

public class Recv \{

private final static String QUEUE\_NAME = "q\_test\_01";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

// 从连接中创建通道

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列

channel.basicConsume(QUEUE\_NAME, true, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[x\] Received '" + message + "'");

5.3.Work模式

5.3.1.图示

一个生产者、2个消费者。

一个消息只能被一个消费者获取。

5.3.2.消费者1

package com.zpc.rabbitmq.work;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv \{

private final static String QUEUE\_NAME = "test\_queue\_work";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 同一时刻服务器只会发一条消息给消费者

//channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，false表示手动返回完成状态，true表示自动

channel.basicConsume(QUEUE\_NAME, true, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[y\] Received '" + message + "'");

//休眠

Thread.sleep(10);

// 返回确认状态，注释掉表示使用自动确认模式

//channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

5.3.3.消费者2

package com.zpc.rabbitmq.work;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv2 \{

private final static String QUEUE\_NAME = "test\_queue\_work";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 同一时刻服务器只会发一条消息给消费者

//channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，false表示手动返回完成状态，true表示自动

channel.basicConsume(QUEUE\_NAME, true, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[x\] Received '" + message + "'");

// 休眠1秒

Thread.sleep(1000);

//下面这行注释掉表示使用自动确认模式

//channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

5.3.4.生产者

向队列中发送100条消息。

package com.zpc.rabbitmq.work;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

public class Send \{

private final static String QUEUE\_NAME = "test\_queue\_work";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

for (int i = 0; i < 100; i++) \{

// 消息内容

String message = "" + i;

channel.basicPublish("", QUEUE\_NAME, null, message.getBytes());

System.out.println(" \[x\] Sent '" + message + "'");

Thread.sleep(i \* 10);

channel.close();

connection.close();

5.3.5.测试

测试结果：

1、消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取。

2、消费者1和消费者2获取到的消息的数量是相同的，一个是消费奇数号消息，一个是偶数。

其实，这样是不合理的，因为消费者1线程停顿的时间短。应该是消费者1要比消费者2获取到的消息多才对。

RabbitMQ 默认将消息顺序发送给下一个消费者，这样，每个消费者会得到相同数量的消息。即轮询(round-robin)分发消息。

怎样才能做到按照每个消费者的能力分配消息呢？联合使用 Qos 和 Acknowledge 就可以做到。

basicQos 方法设置了当前信道最大预获取(prefetch)消息数量为1。消息从队列异步推送给消费者，消费者的 ack 也是异步发送给队列，从队列的视角去看，总是会有一批消息已推送但尚未获得 ack 确认，Qos 的 prefetchCount 参数就是用来限制这批未确认消息数量的。设为1时，队列只有在收到消费者发回的上一条消息 ack 确认后，才会向该消费者发送下一条消息。prefetchCount 的默认值为0，即没有限制，队列会将所有消息尽快发给消费者。

2个概念

轮询分发 ：使用任务队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作，我们可以通过增加工作者(消费者)来解决这一问题，使得系统的伸缩性更加容易。在默认情况下，RabbitMQ将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等，且是提前一次性分配，并非一个一个分配)。平均每个消费者获得相同数量的消息。这种方式分发消息机制称为Round-Robin(轮询)。

公平分发 ：虽然上面的分配法方式也还行，但是有个问题就是：比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将消息发给轮询指定的消费者。

为了解决这个问题，我们使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。

还有一点需要注意，使用公平分发，必须关闭自动应答，改为手动应答。

5.4.Work模式的“能者多劳”

打开上述代码的注释：

// 同一时刻服务器只会发一条消息给消费者

channel.basicQos(1);

//开启这行 表示使用手动确认模式

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

同时改为手动确认：

// 监听队列，false表示手动返回完成状态，true表示自动

channel.basicConsume(QUEUE\_NAME, false, consumer);

测试：

消费者1比消费者2获取的消息更多。

5.5.消息的确认模式

消费者从队列中获取消息，服务端如何知道消息已经被消费呢？

模式1：自动确认

只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费。

模式2：手动确认

消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态。

手动模式：

自动模式：

5.6.订阅模式

5.6.1.图示

解读：

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机，到达队列，实现，一个消息被多个消费者获取的目的

注意：一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费

5.6.2.消息的生产者(看作是后台系统)

向交换机中发送消息。

package com.zpc.rabbitmq.subscribe;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

public class Send \{

private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明exchange

channel.exchangeDeclare(EXCHANGE\_NAME, "fanout");

// 消息内容

String message = "Hello World!";

channel.basicPublish(EXCHANGE\_NAME, "", null, message.getBytes());

System.out.println(" \[x\] Sent '" + message + "'");

channel.close();

connection.close();

注意：消息发送到没有队列绑定的交换机时，消息将丢失，因为，交换机没有存储消息的能力，消息只能存在在队列中。

5.6.3.消费者1(看作是前台系统)

package com.zpc.rabbitmq.subscribe;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv \{

private final static String QUEUE\_NAME = "test\_queue\_work1";

private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 绑定队列到交换机

channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "");

// 同一时刻服务器只会发一条消息给消费者

channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，手动返回完成

channel.basicConsume(QUEUE\_NAME, false, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[Recv\] Received '" + message + "'");

Thread.sleep(10);

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

5.6.4.消费者2(看作是搜索系统)

package com.zpc.rabbitmq.subscribe;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv2 \{

private final static String QUEUE\_NAME = "test\_queue\_work2";

private final static String EXCHANGE\_NAME = "test\_exchange\_fanout";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 绑定队列到交换机

channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "");

// 同一时刻服务器只会发一条消息给消费者

channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，手动返回完成

channel.basicConsume(QUEUE\_NAME, false, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[Recv2\] Received '" + message + "'");

Thread.sleep(10);

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

5.6.5.测试

测试结果：

同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

在管理工具中查看队列和交换机的绑定关系：

5.7.路由模式

5.7.1.图示

5.7.2.生产者

5.7.3.消费者1(假设是前台系统)

5.7.4.消费2(假设是搜索系统)

5.8.主题模式(通配符模式)

5.8.1.图示

同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。

5.8.2.生产者

package com.zpc.rabbitmq.topic;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Send \{

private final static String EXCHANGE\_NAME = "test\_exchange\_topic";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明exchange

channel.exchangeDeclare(EXCHANGE\_NAME, "topic");

// 消息内容

String message = "Hello World!!";

channel.basicPublish(EXCHANGE\_NAME, "routekey.1", null, message.getBytes());

System.out.println(" \[x\] Sent '" + message + "'");

channel.close();

connection.close();

5.8.3.消费者1(前台系统)

package com.zpc.rabbitmq.topic;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv \{

private final static String QUEUE\_NAME = "test\_queue\_topic\_work\_1";

private final static String EXCHANGE\_NAME = "test\_exchange\_topic";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 绑定队列到交换机

channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "routekey.\*");

// 同一时刻服务器只会发一条消息给消费者

channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，手动返回完成

channel.basicConsume(QUEUE\_NAME, false, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[Recv\_x\] Received '" + message + "'");

Thread.sleep(10);

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

5.8.4.消费者2(搜索系统)

package com.zpc.rabbitmq.topic;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.QueueingConsumer;

public class Recv2 \{

private final static String QUEUE\_NAME = "test\_queue\_topic\_work\_2";

private final static String EXCHANGE\_NAME = "test\_exchange\_topic";

public static void main(String\[\] argv) throws Exception \{

// 获取到连接以及mq通道

Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();

// 声明队列

channel.queueDeclare(QUEUE\_NAME, false, false, false, null);

// 绑定队列到交换机

channel.queueBind(QUEUE\_NAME, EXCHANGE\_NAME, "\*.\*");

// 同一时刻服务器只会发一条消息给消费者

channel.basicQos(1);

// 定义队列的消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 监听队列，手动返回完成

channel.basicConsume(QUEUE\_NAME, false, consumer);

// 获取消息

while (true) \{

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" \[Recv2\_x\] Received '" + message + "'");

Thread.sleep(10);

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

6.Spring-Rabbit

6.1.Spring项目

http://spring.io/projects

6.2.简介

6.3.使用

6.3.1.消费者

package com.zpc.rabbitmq.spring;

/\*\*

\* 消费者

\* @author Evan

\*/

public class Foo \{

//具体执行业务的方法

public void listen(String foo) \{

System.out.println("\\n消费者： " + foo + "\\n");

6.3.2.生产者

package com.zpc.rabbitmq.spring;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.context.support.AbstractApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringMain \{

public static void main(final String... args) throws Exception \{

AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(

"classpath:spring/rabbitmq-context.xml");

//RabbitMQ模板

RabbitTemplate template = ctx.getBean(RabbitTemplate.class);

//发送消息

template.convertAndSend("Hello, 鸟鹏!");

Thread.sleep(1000);// 休眠1秒

ctx.destroy(); //容器销毁

6.3.3.配置文件

1、定义连接工厂

host="127.0.0.1" port="5672" username="admin" password="admin"

virtual-host="testhost" />

2、定义模板(可以指定交换机或队列)

3、定义队列、交换机、以及完成队列和交换机的绑定

4、定义监听

5、定义管理，用于管理队列、交换机等：

完整配置文件rabbitmq-context.xml

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xsi:schemaLocation="http://www.springframework.org/schema/rabbit

http://www.springframework.org/s ... ring-rabbit-1.4.xsd

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">

host="127.0.0.1" port="5672" username="admin" password="admin"

virtual-host="testhost" />

6.4.持久化交换机和队列

持久化：将交换机或队列的数据保存到磁盘，服务器宕机或重启之后依然存在。

非持久化：将交换机或队列的数据保存到内存，服务器宕机或重启之后将不存在。

非持久化的性能高于持久化。

如何选择持久化？非持久化？ – 看需求。

7.Spring集成RabbitMQ一个完整案例

创建三个系统A,B,C

A作为生产者，B、C作为消费者(B,C作为web项目启动)

项目下载地址：https://download.csdn.net/download/zpcandzhj/10585077

7.1.在A系统中发送消息到交换机

7.1.1.导入依赖

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

com.zpc

myrabbitA

0.0.1-SNAPSHOT

jar

myrabbit

org.springframework.amqp

spring-rabbit

1.4.0.RELEASE

com.alibaba

fastjson

1.2.47

7.1.2.队列和交换机的绑定关系

实现：

1、在配置文件中将队列和交换机完成绑定

2、可以在管理界面中完成绑定

a)绑定关系如果发生变化，需要修改配置文件，并且服务需要重启

b)管理更加灵活

c)更容易对绑定关系的权限管理，流程管理

本例选择第2种方式

7.1.3.配置

rabbitmq-context.xml

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xsi:schemaLocation="http://www.springframework.org/schema/rabbit

http://www.springframework.org/s ... ring-rabbit-1.4.xsd

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">

host="127.0.0.1" port="5672" username="admin" password="admin"

virtual-host="testhost" />

7.1.4.消息内容

方案：

1、消息内容使用对象做json序列化发送

a)数据大

b)有些数据其他人是可能用不到的

2、发送特定的业务字段，如id、操作类型

7.1.5.实现

生产者MsgSender.java：

package com.zpc.myrabbit;

import com.alibaba.fastjson.JSON;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.context.support.AbstractApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.HashMap;

import java.util.Map;

/\*\*

\* 消息生产者

\*/

public class MsgSender \{

public static void main(String\[\] args) throws Exception \{

AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(

"classpath:spring/rabbitmq-context.xml");

//RabbitMQ模板

RabbitTemplate template = ctx.getBean(RabbitTemplate.class);

String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制

//发送消息

Map msg = new HashMap();

msg.put("type", "1");

msg.put("date", date);

template.convertAndSend("type2", JSON.toJSONString(msg));

Thread.sleep(1000);// 休眠1秒

ctx.destroy(); //容器销毁

7.2.在B系统接收消息

7.2.1.导入依赖

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

com.zpc

myrabbitB

0.0.1-SNAPSHOT

war

myrabbit

4.1.3.RELEASE

1.2.46

com.rabbitmq

amqp-client

3.4.1

org.springframework.amqp

spring-rabbit

1.4.0.RELEASE

org.springframework

spring-webmvc

$\{spring.version\}

com.alibaba

fastjson

1.2.47

$\{project.artifactId\}

org.apache.tomcat.maven

tomcat7-maven-plugin

/testRabbit

UTF-8

8081

7.2.2.配置

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xsi:schemaLocation="http://www.springframework.org/schema/rabbit

http://www.springframework.org/s ... ring-rabbit-1.4.xsd

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">

host="127.0.0.1" port="5672" username="admin" password="admin"

virtual-host="testhost" />

7.2.3.具体处理逻辑

public class Listener \{

//具体执行业务的方法

public void listen(String msg) \{

System.out.println("\\n消费者B开始处理消息： " + msg + "\\n");

7.2.4.在界面管理工具中完成绑定关系

选中定义好的交换机(exchange)

1)direct

2)fanout

3)topic

7.3.在C系统中接收消息

(和B系统配置差不多，无非是Q名和Q对应的处理逻辑变了)

7.3.1.配置

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"

xsi:schemaLocation="http://www.springframework.org/schema/rabbit

http://www.springframework.org/s ... ring-rabbit-1.4.xsd

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">

host="127.0.0.1" port="5672" username="admin" password="admin"

virtual-host="testhost" />

7.3.2.处理业务逻辑

public class Listener \{

//具体执行业务的方法

public void listen(String msg) \{

System.out.println("\\n消费者C开始处理消息： " + msg + "\\n");

7.3.3.在管理工具中绑定队列和交换机

见7.2.4

7.3.4.测试

分别启动B,C两个web应用，然后运行A的MsgSender主方法发送消息，分别测试fanout、direct、topic三种类型

8.Springboot集成RabbitMQ

springboot集成RabbitMQ非常简单，如果只是简单的使用配置非常少，springboot提供了spring-boot-starter-amqp对消息各种支持。

代码下载地址：https://download.csdn.net/download/zpcandzhj/10585077

8.1.简单队列

1、配置pom文件，主要是添加spring-boot-starter-amqp的支持

org.springframework.boot

spring-boot-starter-amqp

2、配置application.properties文件

配置rabbitmq的安装地址、端口以及账户信息

spring.application.name=spirng-boot-rabbitmq

spring.rabbitmq.host=127.0.0.1

spring.rabbitmq.port=5672

spring.rabbitmq.username=admin

spring.rabbitmq.password=admin

3、配置队列

package com.zpc.rabbitmq;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class RabbitConfig \{

@Bean

public Queue queue() \{

return new Queue("q\_hello");

4、发送者

package com.zpc.rabbitmq;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;

import java.util.Date;

@Component

public class HelloSender \{

@Autowired

private AmqpTemplate rabbitTemplate;

public void send() \{

String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制

String context = "hello " + date;

System.out.println("Sender : " + context);

//简单对列的情况下routingKey即为Q名

this.rabbitTemplate.convertAndSend("q\_hello", context);

5、接收者

package com.zpc.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_hello")

public class HelloReceiver \{

@RabbitHandler

public void process(String hello) \{

System.out.println("Receiver  : " + hello);

6、测试

package com.zpc.rabbitmq;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class RabbitMqHelloTest \{

@Autowired

private HelloSender helloSender;

@Test

public void hello() throws Exception \{

helloSender.send();

8.2.多对多使用(Work模式)

注册两个Receiver:

package com.zpc.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_hello")

public class HelloReceiver2 \{

@RabbitHandler

public void process(String hello) \{

System.out.println("Receiver2  : " + hello);

@Test

public void oneToMany() throws Exception \{

for (int i=0;i<100;i++)\{

helloSender.send(i);

Thread.sleep(300);

public void send(int i) \{

String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小时制

String context = "hello " + i + " " + date;

System.out.println("Sender : " + context);

//简单对列的情况下routingKey即为Q名

this.rabbitTemplate.convertAndSend("q\_hello", context);

8.3.Topic Exchange(主题模式)

topic 是RabbitMQ中最灵活的一种方式，可以根据routing\_key自由的绑定不同的队列

首先对topic规则配置，这里使用两个队列(消费者)来演示。

1)配置队列，绑定交换机

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.core.TopicExchange;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class TopicRabbitConfig \{

final static String message = "q\_topic\_message";

final static String messages = "q\_topic\_messages";

@Bean

public Queue queueMessage() \{

return new Queue(TopicRabbitConfig.message);

@Bean

public Queue queueMessages() \{

return new Queue(TopicRabbitConfig.messages);

/\*\*

\* 声明一个Topic类型的交换机

\* @return

\*/

@Bean

TopicExchange exchange() \{

return new TopicExchange("mybootexchange");

/\*\*

\* 绑定Q到交换机,并且指定routingKey

\* @param queueMessage

\* @param exchange

\* @return

\*/

@Bean

Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) \{

return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");

@Bean

Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) \{

return BindingBuilder.bind(queueMessages).to(exchange).with("topic.\#");

2)创建2个消费者

q\_topic\_message 和q\_topic\_messages

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_topic\_message")

public class Receiver1 \{

@RabbitHandler

public void process(String hello) \{

System.out.println("Receiver1  : " + hello);

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_topic\_messages")

public class Receiver2 \{

@RabbitHandler

public void process(String hello) \{

System.out.println("Receiver2 : " + hello);

3)消息发送者(生产者)

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

@Component

public class MsgSender \{

@Autowired

private AmqpTemplate rabbitTemplate;

public void send1() \{

String context = "hi, i am message 1";

System.out.println("Sender : " + context);

this.rabbitTemplate.convertAndSend("mybootexchange", "topic.message", context);

public void send2() \{

String context = "hi, i am messages 2";

System.out.println("Sender : " + context);

this.rabbitTemplate.convertAndSend("mybootexchange", "topic.messages", context);

send1方法会匹配到topic.\#和topic.message，两个Receiver都可以收到消息，发送send2只有topic.\#可以匹配所有只有Receiver2监听到消息。

4)测试

package com.zpc.rabbitmq.topic;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class RabbitTopicTest \{

@Autowired

private MsgSender msgSender;

@Test

public void send1() throws Exception \{

msgSender.send1();

@Test

public void send2() throws Exception \{

msgSender.send2();

8.4.Fanout Exchange(订阅模式)

Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

1)配置队列，绑定交换机

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.core.Binding;

import org.springframework.amqp.core.BindingBuilder;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class FanoutRabbitConfig \{

@Bean

public Queue aMessage() \{

return new Queue("q\_fanout\_A");

@Bean

public Queue bMessage() \{

return new Queue("q\_fanout\_B");

@Bean

public Queue cMessage() \{

return new Queue("q\_fanout\_C");

@Bean

FanoutExchange fanoutExchange() \{

return new FanoutExchange("mybootfanoutExchange");

@Bean

Binding bindingExchangeA(Queue aMessage, FanoutExchange fanoutExchange) \{

return BindingBuilder.bind(aMessage).to(fanoutExchange);

@Bean

Binding bindingExchangeB(Queue bMessage, FanoutExchange fanoutExchange) \{

return BindingBuilder.bind(bMessage).to(fanoutExchange);

@Bean

Binding bindingExchangeC(Queue cMessage, FanoutExchange fanoutExchange) \{

return BindingBuilder.bind(cMessage).to(fanoutExchange);

2)创建3个消费者

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_fanout\_A")

public class ReceiverA \{

@RabbitHandler

public void process(String hello) \{

System.out.println("AReceiver  : " + hello + "/n");

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_fanout\_B")

public class ReceiverB \{

@RabbitHandler

public void process(String hello) \{

System.out.println("BReceiver  : " + hello + "/n");

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

@RabbitListener(queues = "q\_fanout\_C")

public class ReceiverC \{

@RabbitHandler

public void process(String hello) \{

System.out.println("CReceiver  : " + hello + "/n");

3)生产者

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

@Component

public class MsgSenderFanout \{

@Autowired

private AmqpTemplate rabbitTemplate;

public void send() \{

String context = "hi, fanout msg ";

System.out.println("Sender : " + context);

this.rabbitTemplate.convertAndSend("mybootfanoutExchange","", context);

4)测试

package com.zpc.rabbitmq.fanout;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class RabbitFanoutTest \{

@Autowired

private MsgSenderFanout msgSender;

@Test

public void send1() throws Exception \{

msgSender.send();

结果如下，三个消费者都收到消息：

AReceiver : hi, fanout msg

CReceiver : hi, fanout msg

BReceiver : hi, fanout msg

9.总结

使用MQ实现商品数据的同步优势：

1、降低系统间耦合度

2、便于管理数据的同步(数据一致性)

rabbitmq实战 php,【成都校区】RabbitMQ实战教程

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 RabbitMQ实战

相关 RabbitMQ实战

相关 RabbitMQ实战教程

相关 rabbitmq实战 php,【成都校区】RabbitMQ实战教程

相关 SpringBoot RabbitMQ 实战

相关 RabbitMQ 实战教程

相关 docker实战rabbitmq

相关 rabbitmq项目实战

相关 RabbitMQ实战教程(三) : 工作队列

相关 RabbitMQ实战教程(二) : 简单队列