安装Socat安装RabbitMQ

在线安装依赖环境：

yum install gcc

yum install socat

yum install openssl
yum install openssl-devel

mkdir /rabbitmq && cd /rabbitmq

# 上传 erlang-22.0.7-1.el7.x86_64.rpm 安装包上传

# 安装
rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

3. 安装RabbitMQ

cd /rabbitmq

# 上传 rabbitmq-server-3.7.17-1.el7.noarch.rpm 安装包
上传

# 安装
rpm -ivh rabbitmq-server-3.7.17-1.el7.noarch.rpm

4. 开启管理界面及配置

# 开启管理界面
rabbitmq-plugins enable rabbitmq_management

# 配置远程可使用guest登录mq
cd /usr/share/doc/rabbitmq-server-3.7.17

cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

# 修改配置文件
vi /etc/rabbitmq/rabbitmq.config

5. 启动

centos6用这个命令：
/sbin/service rabbitmq-server restart

centos7用这个命令：
systemctl start rabbitmq-server

6. 配置虚拟主机及用户

6.1. 用户角色

角色说明：

1、超级管理员(administrator)

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

2、监控者(monitoring)

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

3、策略制定者(policymaker)

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

4、普通管理者(management)

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

5、其他

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

RabbitMQ消息中间件

MQ优点

1. 应用解耦

1. 异步提速

2. 削锋填谷

MQ缺点

java程序连接RabbitMQ服务

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

</dependency>

</dependencies>

代码--生产者

package com.wwc;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class dome01 {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostconnectionFactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672connectionFactory.setPort(5672);//设置账号和密码 默认guestconnectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟主机名 默认为 /connectionFactory.setVirtualHost("guest");//获取连接通道Connection connection = connectionFactory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue,队列名//boolean durable,是否持久化//boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。//boolean autoDelete,是否自动删除//Map<String, Object> arguments: 其他参数channel.queueDeclare("simple_queue",true,false,false, null);//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/String msg="hello world~~~~3";channel.basicPublish("","simple_queue",null,msg.getBytes());connection.close();}
}

代码-消费者

package com.wwc;import com.rabbitmq.client.*;import java.io.IOException;public class xiaofei {public static void main(String[] args)throws Exception {ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setPassword("guest");factory.setUsername("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("guest");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue 监听得队列名称//autoAck 是否自动确定消息//Consumer callback: 监听到消息后触发的回调函数DefaultConsumer callback = new DefaultConsumer(channel){//一定有消息就会触发该方法//body:表示消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息内容:"+new String(body));}};//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/channel.basicConsume("simple_queue",true,callback);}}

woker模式

生产者

public class dao1 {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostconnectionFactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672connectionFactory.setPort(5672);//设置账号和密码 默认guestconnectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟主机名 默认为 /connectionFactory.setVirtualHost("/");//获取连接通道Connection connection = connectionFactory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue,队列名//boolean durable,是否持久化//boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。//boolean autoDelete,是否自动删除//Map<String, Object> arguments: 其他参数channel.queueDeclare("Work_queue",true,false,false, null);//创建交换机/*  String exchange,交换机的名称48 BuiltinExchangeType type,交换机的种类49 boolean durable:是否持久化50**///发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/for(int i=0;i<10;i++) {String msg = "Worker msg======="+i;channel.basicPublish("","Work_queue",null,msg.getBytes());}connection.close();}
}

消费者

public class xiaofei11 {public static void main(String[] args)throws Exception {ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setPassword("guest");factory.setUsername("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue 监听得队列名称//autoAck 是否自动确定消息//Consumer callback: 监听到消息后触发的回调函数DefaultConsumer callback = new DefaultConsumer(channel){//一定有消息就会触发该方法//body:表示消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息内容:"+new String(body));}};//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/channel.basicConsume("Work_queue",true,callback);}}

生产者

public class dao {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostconnectionFactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672connectionFactory.setPort(5672);//设置账号和密码 默认guestconnectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟主机名 默认为 /connectionFactory.setVirtualHost("/");//获取连接通道Connection connection = connectionFactory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue,队列名//boolean durable,是否持久化//boolean exclusive,（独占）声明队列同一时间只能保 证一个连接，且该队列只有被这一个连接使用。//boolean autoDelete,是否自动删除//Map<String, Object> arguments: 其他参数channel.queueDeclare("work_queue",true,false,false, null);//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/for(int i=0;i<10;i++) {String msg = "Worker msg======="+i;channel.basicPublish("","work_queue",null,msg.getBytes());}connection.close();}
}

消费者

public class xiaofei1 {public static void main(String[] args)throws Exception {ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setPassword("guest");factory.setUsername("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列// 如果该队列名不存在则自动创建，存在则不创建//String queue 监听得队列名称//autoAck 是否自动确定消息//Consumer callback: 监听到消息后触发的回调函数DefaultConsumer callback = new DefaultConsumer(channel){//一定有消息就会触发该方法//body:表示消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息内容:"+new String(body));}};//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没 有交换机""* String routingKey,消息绑定的路由key 如果为简单模 式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/channel.basicConsume("work_queue",true,callback);}}

生产者

public class dao {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("router_queue01", true, false, false, null);channel.queueDeclare("router_queue02", true, false, false, null);//创建交换机/*String exchange,交换机的名称BuiltinExchangeType type,交换机的种类boolean durable:是否持久化*/channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT, true);//交换机和队列绑定/*String queue,队列名 String exchange,交换机名 String routingKey 路由key 如果为发布订阅模式则无需路由key*/channel.queueBind("router_queue01", "direct_exchange", "error");channel.queueBind("router_queue02", "direct_exchange", "error");channel.queueBind("router_queue02", "direct_exchange", "info");channel.queueBind("router_queue02", "direct_exchange", "warning");//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没有交换机""* String routingKey,消息绑定的路由key 如果为简单模式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/String msg = "publisher msg=======2";channel.basicPublish("direct_exchange", "error", null, msg.getBytes());connection.close();
}
}

消费者

public class router {public static void main(String[] args) throws Exception {ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection=factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer callback=new DefaultConsumer(channel){//一定有消息就会触发该方法//body:表示消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息内容:"+new String(body));}};channel.basicConsume("router_queue01",true,callback);}
}

public class dao {public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection = factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//创建队列/*** 如果该队列名不存在则自动创建，存在则不创建* String queue,队列名* boolean durable,是否持久化* boolean exclusive,（独占）声明队列同一时间只能保证一个连接，且该队列只有被这一个连接使用。* boolean autoDelete,是否自动删除* Map<String, Object> arguments: 其他参数*/channel.queueDeclare("topic_queue01", true, false, false, null);channel.queueDeclare("topic_queue02", true, false, false, null);//创建交换机/*String exchange,交换机的名称BuiltinExchangeType type,交换机的种类boolean durable:是否持久化*/channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC, true);//交换机和队列绑定/*String queue,队列名 String exchange,交换机名 String routingKey 路由key 如果为发布订阅模式则无需路由key*/channel.queueBind("topic_queue01", "topic_exchange", "*.orange.*");channel.queueBind("topic_queue02", "topic_exchange", "*.*.rabbit");channel.queueBind("topic_queue02", "topic_exchange", "lazy.#");//发送消息到队列/*** String exchange,把消息发给哪个交换机--简单模式没有交换机""* String routingKey,消息绑定的路由key 如果为简单模式 默认写为队列名称* BasicProperties props, 消息的属性* byte[] body: 消息的内容*/String msg = "routers msg=======2";channel.basicPublish("topic_exchange", "lazy.orange.rabbit", null, msg.getBytes());connection.close();}
}

消费者

public class topic {public static void main(String[] args) throws Exception {ConnectionFactory factory=new ConnectionFactory();//设置rabbitMQ服务器的地址 默认localhostfactory.setHost("192.168.1.130");//设置rabbitMQ的端口号 默认5672factory.setPort(5672);//设置账号和密码 默认guestfactory.setUsername("guest");factory.setPassword("guest");//设置虚拟主机名 默认为 /factory.setVirtualHost("/");//获取连接通道Connection connection=factory.newConnection();//获取channel信道Channel channel = connection.createChannel();//监听队列/*** String queue,监听的队列名称* autoAck:是否自动确认消息* Consumer callback: 监听到消息后触发的回调函数*/DefaultConsumer callback=new DefaultConsumer(channel){//一定有消息就会触发该方法//body:表示消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接受的消息内容:"+new String(body));}};channel.basicConsume("topic_queue01",true,callback);}
}