RabbitMQ安装及配置和使用,消息确认机制

举例:https://blog.csdn.net/qq_35387940/article/details/100514134

Windows下RabbitMQ安装及配置地址: https://blog.csdn.net/zhm3023/article/details/82217222
RabbitMQ(四)订阅模式:https://blog.csdn.net/saytime/article/details/80541328
RabbitMQ(五)路由模式:https://blog.csdn.net/saytime/article/details/80541402
RabbitMQ(六)主题模式:https://blog.csdn.net/saytime/article/details/80541412

 1、简单队列不足:即一个生产者对应一个消费者,一对一的关系,不支持多个消费者

 2、工作队列模式:即一个生产者可以对应多个消费者同时消费,相比简单队列支持多消费者。

3、订阅模式:即一个生产者发送消息给多个消费者,且每个消费者都收到一次,也即是一个消息能够被多个消费者消费。类似于我们订阅同一微信公众号,微信公众号推送图文,我们每个人都能收到一份。

4、路由模式:跟订阅模式类似,只不过在订阅模式的基础上加上了类型,订阅模式是分发到所有绑定到交换机的队列,路由模式只分发到绑定在交换机上面指定路由键的队列。

5、主题模式:跟路由模式类似,只不过路由模式是指定固定的路由键,而主题模式是可以模糊匹配路由键,类似于SQL中=和like的关系。

 //简单模式:// 获取连接.从连接开一个通道,发送消息到队列。 消费者消费队列的消息。

//订阅模式:// 获取连接.从连接开一个通道.声明一个fanout分发交换机.发送消息到交换机。消费者绑定到交换机。消费者消费具体的队列的消息。

/**
 * 生产者
 */
public class Send {
    private static final String EXCHANGE_NAME = "test_exchange_fanout";
public static void main(String[] args) throws IOException, TimeoutException { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 从连接开一个通道 Channel channel = connection.createChannel(); // 声明一个fanout分发交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); String message = "hello, ps"; // 发送消息 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
/**
 * 短信消费者
 */
public class Recv {

    // 短信队列
    private static final String QUEUE_NAME = "test_queue_fanout_sms";
    private static final String EXCHANGE_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 打开通道
        Channel channel = connection.createChannel();
        // 申明要消费的队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说,在接收到该Consumer的ack前,他它不会将新的Message分发给它。
        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);
                System.out.println(" [1] Received '" + message + "'");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(" [1] done ");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        // 消费消息
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

RabbitMQ消息确认机制之事务机制:

1.服务器异常数据丢失问题?

第一种: AMQP 事务处理机制;https://blog.csdn.net/saytime/article/details/80541423

txSelect  将当前channel设置为transaction模式
channel.txCommit  提交当前事务
channel.txRollback  事务回滚
成功就是发送到服务器,失败可以捕获异常再重发。缺点:降低rabbirmq的性能。

第二种: confirm模式 。信号确认或者失败回复:

https://blog.csdn.net/cj337083172/article/details/79730438

1.同步:提交消息后,等待服务端回复后才算确认,概括的简单吧。
2.异步:使用监听方法,异步等待消息回调。

控制方式
开启confirm模式代码 : channel.confirmSelect();
1.同步确认,等待结果方法: channel.waitForConfirms() true-发送成功,false-发送失败
2.异步确认,直接添加一个监听,也能叫做监听模式:
channel.addConfirmListener(new ConfirmListener() {//实现返回处理方法... });

 

 

原文地址:https://www.cnblogs.com/lgg20/p/11692685.html