主要有两种模式：
1、轮询模式的分发：一个消费者一条，按均分配；
2、公平分发：根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少；按劳分配；
一个生产者，两个及以上消费者（同一队列）接收消息即可
// 定义指标 qos=1 finalChannel.basicQos(1); // 改成手动应答 finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);二、公平分发
需要设置指标qos=1并改成手动应答
生产者
package com.test.rabbitmq.work.fair;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
public static void main(String[] args) {
// 所有的中间件技术都是基于tcp/ip协议的基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
// IP port
// 1:创建连接工程
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("106.14.40.217");
connectionFactory.setPort(5672);
connectionFactory.setUsername("admin");
connectionFactory.setPassword("admin");
connectionFactory.setVirtualHost("/");
Connection connection = null;
Channel channel = null;
try {
// 2.创建连接Connection
connection = connectionFactory.newConnection("生产者");
// 3.通过连接获取通道Channel
channel = connection.createChannel();
// 4.准备消息内容
for (int i = 0; i < 20; i++) {
String message = "Hello ,I'm koko"+i;
channel.basicPublish("","queue1",null,message.getBytes());
}
System.out.println("消息发送成功！！！");
} catch (Exception e) {
e.printStackTrace();
} finally {
// 7.关闭通道
if (channel != null && channel.isOpen()){
try {
channel.close();
} catch (Exception e){
e.printStackTrace();;
}
}
// 8.关闭连接
if (connection != null && connection.isOpen()){
try {
connection.close();
} catch (Exception e){
e.printStackTrace();;
}
}
}
}
}
消费者1
package com.test.rabbitmq.work.fair;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Work1 {
public static Runnable runnable = new Runnable() {
@Override
public void run() {
{
// 所有的中间件技术都是基于tcp/ip协议的基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
// IP port
// 1:创建连接工程
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("106.14.40.217");
connectionFactory.setPort(5672);
connectionFactory.setUsername("admin");
connectionFactory.setPassword("admin");
connectionFactory.setVirtualHost("/");
// 2.获取队列的名称
final String queueName = Thread.currentThread().getName();
Connection connection = null;
Channel channel = null;
try {
// 3.创建连接Connection
connection = connectionFactory.newConnection("消费者");
// 4.通过连接获取通道Channel
channel = connection.createChannel();
// channel.queueDeclare("queue_test",false,false,false,null);
// 6.定义接收消息的回调
Channel finalChannel = channel;
// 定义指标 qos=1
finalChannel.basicQos(1);
// 第二个参数设置为false 下面改成手动应答
finalChannel.basicConsume(queueName, false, new DeliverCallback() {
@Override
public void handle(String s, Delivery message) throws IOException {
System.out.println("Work1收到消息是" + new String(message.getBody(), "UTF-8"));
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 改成手动应答
finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
}, new CancelCallback() {
@Override
public void handle(String s) throws IOException {
System.out.println("接收失败了");
}
});
System.out.println("开始接收消息");
System.in.read();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 7.关闭通道
if (channel != null && channel.isOpen()){
try {
channel.close();
} catch (Exception e){
e.printStackTrace();;
}
}
// 8.关闭连接
if (connection != null && connection.isOpen()){
try {
connection.close();
} catch (Exception e){
e.printStackTrace();;
}
}
}
}
}
};
public static void main(String[] args) {
new Thread(runnable,"queue1").start();
}
}
消费者2
package com.test.rabbitmq.work.fair;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Work2 {
public static Runnable runnable = new Runnable() {
@Override
public void run() {
{
// 所有的中间件技术都是基于tcp/ip协议的基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
// IP port
// 1:创建连接工程
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("106.14.40.217");
connectionFactory.setPort(5672);
connectionFactory.setUsername("admin");
connectionFactory.setPassword("admin");
connectionFactory.setVirtualHost("/");
// 2.获取队列的名称
final String queueName = Thread.currentThread().getName();
Connection connection = null;
Channel channel = null;
try {
// 3.创建连接Connection
connection = connectionFactory.newConnection("消费者");
// 4.通过连接获取通道Channel
channel = connection.createChannel();
// channel.queueDeclare("queue_test",false,false,false,null);
// 6.定义接收消息的回调
Channel finalChannel = channel;
// 定义指标 qos=1
finalChannel.basicQos(1);
// 第二个参数设置为false 下面改成手动应答
finalChannel.basicConsume(queueName, false, new DeliverCallback() {
@Override
public void handle(String s, Delivery message) throws IOException {
System.out.println("Work2收到消息是" + new String(message.getBody(), "UTF-8"));
// 改成手动应答
finalChannel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
}, new CancelCallback() {
@Override
public void handle(String s) throws IOException {
System.out.println("接收失败了");
}
});
System.out.println("开始接收消息");
System.in.read();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 7.关闭通道
if (channel != null && channel.isOpen()){
try {
channel.close();
} catch (Exception e){
e.printStackTrace();;
}
}
// 8.关闭连接
if (connection != null && connection.isOpen()){
try {
connection.close();
} catch (Exception e){
e.printStackTrace();;
}
}
}
}
}
};
public static void main(String[] args) {
new Thread(runnable,"queue1").start();
}
}
