下载安装的环境为centos7，并未写明如何上传文件到Linux虚拟机中，相信能学到MQ的朋友们都应该多少了解过如何上传文件，虽然我本人习惯用lrzsz的方式。java整合使用rabbitmq分别有非注解和注解两种风格。
RabbitMQ安装 安装依赖环境在线安装依赖环境：
yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz安装Erlang
上传
erlang-18.3-1.el7.centos.x86_64.rpm
socat-1.7.3.2-1.1.el7.x86_64.rpm
rabbitmq-server-3.6.5-1.noarch.rpm
# 安装 rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm安装RabbitMQ
# 安装 rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm # 安装 rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm开启管理界面及配置
# 开启管理界面 rabbitmq-plugins enable rabbitmq_management # 修改默认配置信息 vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app # 比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest启动
service rabbitmq-server start # 启动服务 service rabbitmq-server stop # 停止服务 service rabbitmq-server restart # 重启服务
- 设置配置文件
cd /usr/share/doc/rabbitmq-server-3.6.5/ cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config配置虚拟主机及用户
RabbitMQ在安装好后，可以访问http://ip地址:15672 ；其自带了guest/guest的用户名和密码；如果需要创建自定义用户；那么也可以登录管理界面后，如下操作：
特征：多个消费端共同消费同一个队列中的消息
应用：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
小结：
- 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。
2. Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，
只需要有一个节点成功发送即可。
生产者测试代码：
package com;
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 helloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("192.168.96.133");
factory.setPort(5672);
factory.setVirtualHost("/test");
factory.setUsername("hrd");
factory.setPassword("hrd");
//3.创建连接
Connection connection = factory.newConnection();
//4.设置channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("hello_world",true,false,false,null);
//需要传递的消息内容
String body = "hello rabbitmq~~~";
//6.发送消息
channel.basicPublish("","hello_world",null,body.getBytes());
//7.释放资源
channel.close();
connection.close();
}
}
消费者测试代码：
package com;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class helloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("192.168.96.133");
factory.setPort(5672);
factory.setVirtualHost("/test");
factory.setUsername("hrd");
factory.setPassword("hrd");
//3.创建连接
Connection connection = factory.newConnection();
//4.设置channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("hello_world",true,false,false,null);
//6.接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("consumerTag："+consumerTag);
System.out.println("Exchange："+envelope.getExchange());
System.out.println("RoutingKey："+envelope.getRoutingKey());
System.out.println("properties："+properties);
System.out.println("body："+new String(body));
}
};
channel.basicConsume("hello_world",true,consumer);
}
}
订阅模式Fanout
特征：多了一个 Exchange 角色
应用：
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合
路由规则的队列，那么消息会丢失！
Exchange有常见以下3种类型：
➢ Fanout：广播，将消息交给所有绑定到交换机的队列
➢ Direct：定向，把消息交给符合指定routing key 的队列
➢ Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
小节：
交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
发布订阅模式与工作队列模式的区别：
-
工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机
-
发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用
默认交换机)
-
发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队
列绑 定到默认的交换机
生产者测试代码：
package com;
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
public class pubSub {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("192.168.96.133");
factory.setPort(5672);
factory.setVirtualHost("/test");
factory.setUsername("hrd");
factory.setPassword("hrd");
//3.创建连接
Connection connection = factory.newConnection();
//4.设置channel
Channel channel = connection.createChannel();
//5. 创建交换机
String exchangeName = "test_fanout";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
//6.创建队列Queue
//如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7. 绑定队列和交换机
channel.queueBind(queue1Name,exchangeName,"");
channel.queueBind(queue2Name,exchangeName,"");
//8.发送消息
//需要传递的消息内容
String body = "hello rabbitmq~~~";
channel.basicPublish(exchangeName,"",null,body.getBytes());
//9.释放资源
channel.close();
connection.close();
}
}
生产者测试截图：
消费者测试代码：
package com;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class pubSub2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("192.168.96.133");
factory.setPort(5672);
factory.setVirtualHost("/test");
factory.setUsername("hrd");
factory.setPassword("hrd");
//3.创建连接
Connection connection = factory.newConnection();
//4.设置channel
Channel channel = connection.createChannel();
//5.创建队列Queue
//如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
//由于先执行生产者，则已经创建，不需要重复
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
// channel.queueDeclare(queue1Name,true,false,false,null);
//6.接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag："+consumerTag);
// System.out.println("Exchange："+envelope.getExchange());
// System.out.println("RoutingKey："+envelope.getRoutingKey());
// System.out.println("properties："+properties);
System.out.println("在数据库中：body："+new String(body));
}
};
channel.basicConsume(queue2Name,true,consumer);
}
}
消费者测试截图：
说明：
-
队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
-
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
-
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息
小结：
Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。
截图为原生非注解代码
以下是使用注解生产者
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testDirect(){
rabbitTemplate.convertAndSend("directs","error","error 的日志信息");
}
使用注解消费者
@Component
public class DirectCustomer {
@RabbitListener(bindings ={
@QueueBinding(
value = @Queue(), key={"info","error"},
exchange = @Exchange(type = "direct",name="directs")
)})
public void receive1(String message){
System.out.println("message1 = " + message);
}
@RabbitListener(bindings ={
@QueueBinding(
value = @Queue(), key={"error"},
exchange = @Exchange(type = "direct",name="directs")
)})
public void receive2(String message){
System.out.println("message2 = " + message);
}
}
通配符模式Topics
说明：
-
Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
-
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
-
通配符规则：
#匹配一个或多个词， 匹配不多不少恰好1个词*
例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert
生产者
@Autowired
private RabbitTemplate rabbitTemplate;
//topic
@Test
public void testTopic(){
rabbitTemplate.convertAndSend("topics","user.save.findAll","user.save.findAll 的消息");
}
消费者
@Component
public class TopCustomer {
@RabbitListener(bindings = {
@QueueBinding(
value = @Queue, key = {"user.*"},
exchange = @Exchange(type = "topic",name = "topics")
)
})
public void receive1(String message){
System.out.println("message1 = " + message);
}
@RabbitListener(bindings = {
@QueueBinding(
value = @Queue, key = {"user.#"},
exchange = @Exchange(type = "topic",name = "topics")
)
})
public void receive2(String message){
System.out.println("message2 = " + message);
}
}
Spring Boot整合RabbitMQ
生产者
配置文件：application.yml
spring:
rabbitmq:
host: 192.168.96.133
port: 5672
username: hrd
password: hrd
virtual-host: /test
配置@Configuration类：config.rabbitMQConfig.java
package com.example.producerboot.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class rabbitMQConfig {
public static final String EXCHANGE_NAME="boot_topic_exchange";
public static final String QUEUE_NAME="boot_topic_change";
// 1.交换机
@Bean("bootExchange")
public Exchange bootExchange(){
return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
}
// 2.队列
@Bean("bootQueue")
public Queue bootQueue(){
return QueueBuilder.durable(QUEUE_NAME).build();
}
// 3.队列和交换机绑定 Bingding
// 知道哪个队列->知道哪个交换机->知道哪个routing key
// @Qualifier注入对应名字的队列和交换机
@Bean
public Binding bingQueueExchange(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchange") Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
}
测试Test类：ProducerBootApplicationTests.java
RabbitTemplate类模板，简化操作，使用时需要注入
package com.example.producerboot;
import com.example.producerboot.config.rabbitMQConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class ProducerBootApplicationTests {
// 1.注入rabbitTemplate
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads() {
rabbitTemplate.convertAndSend(rabbitMQConfig.EXCHANGE_NAME,"boot.haha","boot mq hello~");
}
}
消费者
maven中pom.xml文件依赖和application.yml配置文件内容与生产者一样
消费者测试代码：
package com.example.consumerboot;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class rabbitMQListener {
@RabbitListener(queues = "boot_topic_change")
public void listenerQueue(Message message){
System.out.println(message);
}
}
高级特性
待续后期系统学习整理~
