当有多个消费者同时收取消息，且每个消费者在接收消息的同时，还要做其它的事情，且会消耗很长的时间，在此过程中可能会出现一些意外，比如消息接收到一半的时候，一个消费者宕掉了，这时候就要使用消息接收确认机制，可以让其它的消费者再次执行刚才宕掉的消费者没有完成的事情。另外，在默认情况下，我们创建的消息队列以及存放在队列里面的消息，都是非持久化的，也就是说当RabbitMQ宕掉了或者是重启了，创建的消息队列以及消息...
消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢——消息持久化。 为了保证RabbitMQ在退出或者crash等异常情况下数据没有丢失，需要将queue，exchange和Message都持久化。queue的持久化queue的持久化是通过durable=true来实现的。 一般程序中这么使用：Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("queue.persistent...
一、队列持久化 声明队列queue_declare方法的原型 ：channel.queue_declare(queue=, passive=False, durable=False,exclusive=False, auto_delete=False,arguments=None):queue: 队列名称 durable： 是否持久化, 队列的声明默认是False，即存放到内存中的，如果rabbitmq重启会丢失。如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库。 exclusive：是否排外的，默认为F...
$ename = "ename14"; $qname = "qname14"; $con = new AMQPConnection(); $con->connect(); $ch = new AMQPChannel($con); $ex = new AMQPExchange($ch); $ex->setName($ename); $ex->setType(AMQP_EX_TYPE_DIRECT); $ex->setFlags(2); $ex->declare(); $q = new AMQPQueue($ch); $q->setName($qname); $q->setFlags(2); $q->declare(); $q->bind($ex->getName(),$qname); $ex->publish...
数据库 我使用的是 PostgreSQL。 使用的驱动是 github.com/lib/pq 这个网址 https://pkg.go.dev/github.com/lib/pq 是官方文档。 创建数据表 打开 PostgreSQL 的管理工具 pgAdmin 4（web 版），创建一个数据库： 数据库名叫 distributed： 创建数据表 首先创建 sensor 表，里面存放传感器信息： 列如下，其中id 是自增主键： 里面需要有传感器的数据，名称要和批处理文件中传感器的名称要一致。 另一个...
虽然有了消息反馈机制，但是如果rabbitmq自身挂掉的话，那么任务还是会丢失。所以需要将任务持久化存储起来。声明持久化存储： 重新定义一个队列 1 channel.queue_declare(queue=task_queue, durable=True) 在发送任务的时候，用delivery_mode=2来标记任务为持久化存储：1 channel.basic_publish(exchange=, 2 routing_key=task_queue, 3 body=message, 4 ...
1 消息确认在一些场合，如转账、付费时每一条消息都必须保证成功的被处理。AMQP是金融级的消息队列协议，有很高的可靠性，这里介绍在使用RabbitMQ时怎么保证消息被成功处理的。消息确认可以分为两种：一种是生产者发送消息到Broke时，Broker给生产者发送确认回执，用于告诉生产者消息已被成功发送到Broker；一种是消费者接收到Broker发送的消息时，消费者给Broker发送确认回执，用于告诉消费者消息已成功被消费者接收。下边分别介绍...