MQ是什么
官网
RabbitMQ是一个开源的,在AMQP基础上完整的,可复用的企业消息系统.
消息队列(Message Queue),从字面上看,本质是队列,FIFO先进先出,只不过队列中存放内容是message而已. 其主要用途是在不同进程Process/线程Thread之间通信
支持主流的操作系统,Linux,weindows,MacOX等
多种开发语言支持,java\Python\Ruby\ .NET\PHP\C\C++\node.js等等
市面主流MQ有:
RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里开源的RocketMQ。本文主要介绍RabbitMq。
MQ做什么
- 不同进程(process)之间传递消息时,两个进程之间耦合程度过高,改动一个进程,引发必须修改另一个进程,为了隔离两个进程,在两个进程间抽离出一个模块,两进程之间传递的消息,都必须通过消息队列来传递,单独修改某一个进程,不会影响另一个;
- 不同进程之间传递消息时,为了实现标准化,将消息的格式规范化了,并且某一个进程接收的消息太多,一下子无法处理完,并且也有先后顺序,必须对收到的消息进行排队,因此诞生了事实上的消息队列
MQ学习
RabbitMQ概念:
- RabbitMQ的内部对象,用于存储消息
- producer生产者,用于服务端发送消息
- consumer消费者,用于客户端读取消息
- queue队列,用于做一个消息端发送消息到某个容器,这个容器的规则是先进先出的. 如服务发送消息到queue,然后消费者客户端就在queue上回去消息
- routing_key, 在Exchange中绑定Queue中需要指定的RoutingKey,用于匹配消息转发到那个Queue中
- exchange 交换机, AMQP 协议中的核心思想就是生产者和消费者的解耦，生产者从不直接将消息发送给队列。生产者通常不知道是否一个消息会被发送到队列中，只是将消息发送到一个交换机。先由 Exchange 来接收，然后 Exchange 按照特定的策略转发到 Queue 进行存储。Exchange 就类似于一个交换机，将各个消息分发到相应的队列中。
- 交换机有4种策略
- Fanout Exchange
Fanout Exchange会忽略RouteingKeyd的设置,直接将Message广播到所有绑定的Queue中 - Direct Exchange
是RabbitMQ默认的Exchange,完全根据RoutingKey来路由消息. 设置Exchange和Queue的Binding是需要指定RoutingKey,发消息时指定RoutingKey,消息到交换机判断RoutingKey之后会转发到对应的Queue中 - Topic Exchange
路由规则跟Direct策略一样,但是Topic策略支持模糊匹配来确定匹配到Queue. 规则: *和#通配符, *表示匹配一个单词, #则表示匹配没有或者多个单词 - Headers Exchange
该策略会忽略RoutingKey而根据消息中的Headers和创建绑定关系指定的Arguments来匹配决定路由到哪些Queue
- Fanout Exchange
- 交换机有4种策略
Rabbitmq模式6大模式
image
主要讲解Topic模式使用,其他模式都是弟弟!! 哈哈
依赖
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.5.1</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.0.RELEASE</version>
</dependency>
producer端,发送java和mysql的信息
import java.util.UUID;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
//发送消息端
public class SendLogTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws Exception {
// 创建连接和通道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//定义RoutingKey
String[] routing_keys = new String[] { "java.info", "java.warning",
"mysql.info", "mysql.critical" };
for (String routing_key : routing_keys)
{
//发送4条不同绑定键的消息
String msg = UUID.randomUUID().toString();
//发送消息
channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg
.getBytes());
System.out.println(" [x] Sent routingKey = "+routing_key+" ,msg = " + msg + ".");
}
channel.close();
connection.close();
}
}
comsumer1(接口java的所有消息)
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//接收java.*消息
public class Comsumer1 {
ConnectionFactory factory = new ConnectionFactory();
factory.setVirtualHost("/");
factory.setHost("127.0.0.1");
factory.setPort(AMQP.PROTOCOL.PORT); // 5672
factory.setUsername("admin");
factory.setPassword("admin");
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
String EXCHANGE_NAME = "topic_logs";
String QUEUE_NAME = "javaQueue";
String key = "java.*"
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
// 只允许一次执行一个消息,ack未提交会则不会接受到下一条消息,配置了自动应对的情况下该配置不生效
channel.basicQos(1);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(message);
//以上的信息处理成功就手动提交,false为确认提交
channel.basicAck(envelope.getDeliveryTag,false);
}
};
//false为手动提交
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
comsumer2(接口mysql.info信息)
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//接收java.*消息
public class Comsumer2 {
ConnectionFactory factory = new ConnectionFactory();
factory.setVirtualHost("/");
factory.setHost("127.0.0.1");
factory.setPort(AMQP.PROTOCOL.PORT); // 5672
factory.setUsername("admin");
factory.setPassword("admin");
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
String EXCHANGE_NAME = "topic_logs";
String QUEUE_NAME = "javaQueue";
//精确匹配
String key = "mysql.info"
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key");
// 只允许一次执行一个消息,ack未提交会则不会接受到下一条消息,配置了自动应对的情况下该配置不生效
channel.basicQos(1);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(message);
//以上的信息处理成功就手动提交,false为确认提交
channel.basicAck(envelope.getDeliveryTag,false);
}
};
//false为手动提交
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
