标签：服务 其他 之一 通道 tco ring rabbit one message
工作队列模式为一个生产者对应多个消费者，但是只有一个消费者获得消息，即一个队列被多个消费者监听，但一条消息只能被其中的一个消费者获取
代码如下：
生产者代码：
public class WorkSend {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//获取连接
Connection connection = ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
//声明通道
Channel channel = connection.createChannel();
//创建队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicPublish("", "hello", null, "First.".getBytes());
channel.basicPublish("", "hello", null, "Secode..".getBytes());
channel.basicPublish("", "hello", null, "Third....".getBytes());
channel.basicPublish("", "hello", null, "Fourth....".getBytes());
channel.basicPublish("", "hello", null, "Fifth.....".getBytes());
//6、关闭通道和连接
channel.close();
connection.close();
}
}
? 消费者代码
public class WorkRecv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//获取连接
Connection connection = ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
//声明通道
Channel channel = connection.createChannel();
//声明队列队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
try {
doWork(message);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println(" [x] Done");
//channel.basicAck();
}
};
boolean autoAck = true; // acknowledgment is covered below
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
});
// DeliverCallback deliverCallback = new DeliverCallback(){
// @Override
// public void handle(String consumerTag, Delivery delivery) throws IOException {
// String message = new String(delivery.getBody(), "UTF-8");
// System.out.println(" [x] Received '" + message + "'");
// }
// };
//
// channel.basicConsume(QUEUE_NAME, true, deliverCallback, new CancelCallback(){
// @Override
// public void handle(String consumerTag) throws IOException {
//
// }
// });
}
private static void doWork(String task) throws InterruptedException {
for (char ch : task.toCharArray()) {
if (ch == '.') Thread.sleep(1000);
}
}
}
1 生产者将消息交个交换机
2 交换机交给绑定的队列
3 队列由多个消费者同时监听,只有其中一个能够获取这一条消息,形成了资源的争抢,谁的资源空闲大,争抢到的可能越大;
Round-robin dispatching(轮询分发)
使用任务队列的优点之一是能够轻松并行工作。如果我们这里积压了很多的消息，我们可以增加work的并行度，这样就可以轻松扩展。
默认情况下，RabbitMQ将每个消息依次发送给下一个消费者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为轮询。
Message acknowledgment(消息确认)
使用我们当前的代码，RabbitMQ一旦向消费者发送了一条消息，便立即将其标记为删除。在这种情况下，如果我们kill掉一个worker，我们将丢失正在处理的消息。并且还将丢失所有发送给该特定worker但尚未处理的消息。
为了确保消息永不丢失，RabbitMQ支持 消息确认。消费者发送回一个确认（告知），告知RabbitMQ特定的消息已被接收并处理，并且RabbitMQ可以自由删除它。
如果消费者死了（其通道已关闭，连接已关闭或TCP连接丢失）而没有发送确认，RabbitMQ将了解消息未完全处理，并将重新排队。如果同时有其他消费者在线，它将很快将其重新分发给另一个消费者。这样，您可以确保即使worker偶尔死亡也不会丢失任何消息。
没有任何消息超时；消费者死亡时，RabbitMQ将重新传递消息。即使处理一条消息花费非常非常长的时间也没关系。
默认情况下，手动消息确认处于打开状态。我们通过autoAck = false 标志显式关闭了它们。在消息完成投递的时候,手动确认消息投递成功。
消费者代码修改如下：
public class ManWorkRecv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
//获取连接
Connection connection = ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
//声明通道
Channel channel = connection.createChannel();
//声明队列队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
try {
doWork(message);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println(" [x] Done");
//手动确认消息已经成功投递
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
boolean autoAck = false; //设置消息手动确认
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {
});
}
private static void doWork(String task) throws InterruptedException {
for (char ch : task.toCharArray()) {
if (ch == '.') Thread.sleep(1000);
}
}
}
? 上述代码可以确保我们在kill掉一个消费者的情况下，消息不会丢失。
Message durability(消息持久化)
前面已经讲了即使一个消费者退出,消息不会丢失，但是如果RabbitMQ服务退出时，队列和消息仍然会丢失,这是因为 默认队列和消息都是放在内存中的 。为保证消息和队列不丢失，需要把队列和消息设置为持久化
确保RabbitMQ永远不会丢失我们的队列。声明持久化代码如下：
boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);由于RabbitMQ已经存在一个hello队列，并且RabbitMQ不支持对已经存在的队列进行参数修改，所以需要我们删除之前创建的队列或者重新创建一个队列。
其次，我们需要保证消息的持久化，消息持久化设置如下：
channel.basicPublish("", "hello", MessageProperties.PERSISTENT_TEXT_PLAIN, "Fifth.....".getBytes());这样，我们就可以保证RabbitMQ服务宕机的情况下，消息和队列都不会丢失。
Fair dispatching(公平分发)
? RabbitMQ在进行消息分发的时候，可以设置一次分发给某一个消费者多少条消息，如在消费者端设置prefetchCount=1;如下代码
int prefetchCount = 1 ;
channel.basicQos（prefetchCount）;该设置表示，RabbitMQ一次分发给消费者一条消息，在消费者处理并确认上一条消息之前，不会再给这个消费者发送一条新消息，而会将其分发给其他的消费者
SpringBoot实现：
@SpringBootApplication
@EnableScheduling
public class RabbitAmqpTutorialsApplication {
public static void main(String[] args) throws Exception {
SpringApplication.run(RabbitAmqpTutorialsApplication.class, args);
}
} @Configuration
public class Tut2Config {
@Bean
public Queue hello() {
return new Queue("hello");
}
/**
* 这里是启动两个消费者
*/
private static class ReceiverConfig{
@Bean
public Tut2Receiver receiver1(){
return new Tut2Receiver(1);
}
@Bean
public Tut2Receiver receiver2(){
return new Tut2Receiver(2);
}
}
@Bean
public Tut2Sender sender() {
return new Tut2Sender();
}
}
//监听队列
@RabbitListener(queues = "hello")
public class Tut2Receiver {
private final int instance;
public Tut2Receiver(int i) {
this.instance = i;
}
@RabbitHandler
public void receive(String in) throws InterruptedException {
StopWatch watch = new StopWatch();
watch.start();
System.out.println("instance " + this.instance +
" [x] Received '" + in + "'");
doWork(in);
watch.stop();
System.out.println("instance " + this.instance +
" [x] Done in " + watch.getTotalTimeSeconds() + "s");
}
private void doWork(String in) throws InterruptedException {
for (char ch : in.toCharArray()) {
if (ch == '.') {
Thread.sleep(1000);
}
}
}
}public class Tut2Sender {
@Autowired
private RabbitTemplate template;
//队列
@Autowired
private Queue queue;
AtomicInteger dots = new AtomicInteger(0);
AtomicInteger count = new AtomicInteger(0);
/**
* 定时向队列hello发送消息
*/
@Scheduled(fixedDelay = 1000, initialDelay = 500)
public void send() {
StringBuilder builder = new StringBuilder("Hello");
if (dots.incrementAndGet() == 3) {
dots.set(1);
}
for (int i = 0; i < dots.get(); i++) {
builder.append('.');
}
builder.append(count.incrementAndGet());
String message = builder.toString();
//向队列中发送消息
template.convertAndSend(queue.getName(), message);
System.out.println(" [x] Sent '" + message + "'");
}
}标签：服务 其他 之一 通道 tco ring rabbit one message
原文地址：https://www.cnblogs.com/haizhilangzi/p/12301722.html
