JMS组件：activemq（慢）
AMQP组件（advance message queue protocol）：rabbitmq和kafka
一.、消息队列解决了什么问题？
异步处理
应用解耦
流量削锋
日志处理
二、rabbitmq安装与配置
三、java操作rabbitmq
1. simple 简单队列
2. work queues 工作队列 公平分发 轮询分发
3. publish/subscribe 发布于订阅
4. routing 路由选择 通配符模式
5. topics 主题
6. 手动和自动确认消息
7. 队列的持久化和非持久化
8. rabbitmq的延迟队列
四、spring AMQP spring-rabbitmq
五、场景demo mq实现搜索引擎DIH增量
六、场景demo 未支付订单30分钟，取消
七、大数据应用 类似百度统计 cnzz架构 消息队列
一、简单队列
ConnectionUtils.java
public class ConnectionUtils {
public static Connection getConnection() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setPort();
factory.setVirtualHost("/vhost_mmr");
factory.setUsername("cxx");
factory.setPassword("cxx");
return factory.newConnection();
}
}
Send.java
/**
* 生产者发送消息
*/
public class Send { private static final String QUEUE_NAME = "test_simple_queue"; public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection(); Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null); String msg = "hello simple!!!!!!!!!!!";
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("--send msg；" + msg);
channel.close();
connection.close();
}
}
Receive.java
/**
* 消费者获取消息
*/
public class Receive { private static final String QUEUE_NAME = "test_simple_queue"; public static void main(String[] args) throws IOException, TimeoutException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//创建通道
Channel channel = connection.createChannel();
//队列声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null); DefaultConsumer 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);
String msg = new String(body, "utf-8");
System.out.println("new api recv:" + msg);
}
}; //监听队列
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
二、工作队列
2.1 轮询分发
Send.java
/**
* \---c1
* p---Queue----\
* \---c2
*/
public class Send {
private static final String QUEUE_NAME = "test_work_queue"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//获取channel
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 0; i < 50; i++) {
String msg = "hello" + i; System.out.println("[WQ ] send:" + msg);
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
Thread.sleep(i * 20);
}
channel.close();
connection.close();
}
}
Receive1.java
public class Receive1 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[1] Recv msg :" + msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[1] done");
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
Receive2.java
public class Receive2 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[2] Recv msg :" + msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done");
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
现象：消费者1和消费者2处理的消息是一样的
消费者1偶数
消费者1奇数
这种方式叫做轮询分发（round-robin）,结果就是不管谁忙活着谁清闲，都不会多给一个消息，任意消息总是你一个，我一个
2.2 公平分发（fair dipatch）
Send.java
/**
* \---c1
* p---Queue----\
* \---c2
*/
public class Send {
private static final String QUEUE_NAME = "test_work_queue"; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//获取链接
Connection connection = ConnectionUtils.getConnection();
//获取channel
Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/**
* 每个消费者发送消费之前，消息队列不发送下一个消息到消费者，一次只处理一个消息
* 限制发送给同一个消费者不得超过一个消息
*/
int prefetchCount = 1;
channel.basicQos(prefetchCount); for (int i = 0; i < 50; i++) {
String msg = "hello" + i; System.out.println("[WQ ] send:" + msg);
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
Thread.sleep(i * 5);
}
channel.close();
connection.close();
}
}
Receive1.java
public class Receive1 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//保证一次只发送一个
channel.basicQos(1);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[1] Recv msg :" + msg);
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[1] done");
//手动回执
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
Receive2.java
public class Receive2 {
public static final String QUEUE_NAME = "test_work_queue";
public static void main(String[] args) throws IOException, TimeoutException {
//创建链接
Connection connection = ConnectionUtils.getConnection();
//创建频道
final Channel channel = connection.createChannel();
//声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//保证一次只发送一个
channel.basicQos(1);
//定义一个消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[2] Recv msg :" + msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done");
//手动回执
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
}
}
现象：消费者2处理的消息比消费者1多，能者多劳
三、消息与应答ack与消息持久化durable
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
boolean autoAck = true;(自动确认模式)一旦rabbitmq将消息分发给消费者，就会从内存中删除
这种情况下，如果杀死正在执行的消费者，就会丢失正在处理的消息
boolnea autoAck = false;(手动模式)，如果有一个消费者挂掉，就会交付给其他消费者，
rabbitmq支持消息应答，消费者发送一个应答，告诉rabbitmq这个消息我已经处理完成，你可以删除了，然后rabbitmq就删除
内存中的消息
消息应答模式是打开的，false
Message acknowkedgment
消息持久化
boolean durable = false
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);