AI 助理
备案控制台
开发者社区 云原生 云消息队列 文章 正文

RabbitMQ消息队列学习笔记

2019-04-20 1974
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 初次使用AMQP的过程中，总是容易被AMQP支持的消息模型绕晕，这里结合官方的教程，对AMQP的消息模型做一个简要总结，供参考。目前官方给出了六种消息发送/接收模型，这里主要介绍前五种消息模型。

概述

初次使用AMQP的过程中，总是容易被AMQP支持的消息模型绕晕，这里结合官方的教程，对AMQP的消息模型做一个简要总结，供参考。目前官方给出了六种消息发送/接收模型，这里主要介绍前五种消息模型。

消息模型

1、Hello World

简单模式就是生产者将消息发送到队列、消费者从队列中获取消息。一条消息对应一个消费者。

_

示例代码说明：

测试使用的是阿里云的AMQP消息队列服务，具体的代码配置过程可以参考阿里云官方链接。

工具类

import AMQP.AliyunCredentialsProvider;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ConnectionUtil {

  public static Connection getConnection() throws Exception{
      // 初始化参数设置
      String AccessKey= "********";
      String SecretKey = "********";
      Long Uid = ********16617278L;
      String VhostName = "********";
      String host = "********16617278.mq-amqp.cn-hangzhou-a.aliyuncs.com";

      // 定义连接工厂
      ConnectionFactory connectionFactory = new ConnectionFactory();
      // 设置服务地址
      connectionFactory.setHost(host);
      // 端口
      connectionFactory.setPort(5672);
      // 设置用户名、密码、vhost
      connectionFactory.setCredentialsProvider(new AliyunCredentialsProvider(AccessKey,SecretKey,Uid));
      connectionFactory.setAutomaticRecoveryEnabled(true);
      connectionFactory.setNetworkRecoveryInterval(5000);
      connectionFactory.setVirtualHost(VhostName);

      // 通过工厂获取连接对象
      Connection connection = connectionFactory.newConnection();
      return connection;
  }
}

发送端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

// hello world 单个消费者和接收者
public class Send {

    private final static String Queue_name = "helloDemo";
    public static void main(String[] args) throws Exception {
        // 获取连接及mq通道
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(Queue_name,false,false,false,null);
        //消息内容
        String message = "Hello World!";
        // 1、交换机，此处无   2、发送到那个队列 3、属性  4、消息内容
        channel.basicPublish("",Queue_name,null,message.getBytes());

        System.out.println("发送数据：" + message);

        // 关闭连接
        channel.close();
        connection.close();
    }
}

消费端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Receiver {
    private final static String Queue_name = "helloDemo";
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel =  connection.createChannel();
        
        // 开始消费消息
        channel.basicConsume(Queue_name, false, "ConsumerTag", new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理
                System.out.println("message receive: ");
                System.out.println("Received： " + new String(body, "UTF-8") + ", deliveryTag: " + envelope.getDeliveryTag());
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
        Thread.sleep(100000);
        channel.close();
        connection.close();
    }
}
2、Work Queues

一条消息可以被多个消费者尝试接收，最终只有一个消费者能够获取到消息。

_

发送端示例代码


import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

// 1：N  消费者各自接收消息
public class Sender {

    private final static String queueName = "workQueue";
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(queueName,false,false,false,null);
        for (int i = 0; i < 100; i++) {

            String message = "workqueues message " + i;
            channel.basicPublish("",queueName,null,message.getBytes());
            System.out.println("发送消息： " + message);

            Thread.sleep(10);//休眠
        }
        // 关闭连接
        channel.close();
        connection.close();
    }
}

消费端示例代码1

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Receiver1 {

    private final static String queueName = "workQueue";
    public static void main(String[] args) throws Exception{

        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(queueName,false,false,false,null);
        channel.basicQos(1);//告诉服务器，在没有确认当前消息完成之前，不要给我发新的消息。

        DefaultConsumer consumer = new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理
                System.out.println("message receive1: ");
                System.out.println("Received1： " + new String(body, "UTF-8") + ", deliveryTag: " + envelope.getDeliveryTag());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);// 参数2 false为确认收到消息， true为拒绝收到消息
            }
        };
        channel.basicConsume(queueName,false,consumer);// 参数2 手动确认，代表我们收到消息后需要手动确认告诉服务器我们收到消息了
    }
}

消费端示例代码2

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Receiver2 {

    private final static String queueName = "workQueue";
    public static void main(String[] args) throws Exception{

        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(queueName,false,false,false,null);
        channel.basicQos(1);//告诉服务器，在没有确认当前消息完成之前，不要给我发新的消息。

        DefaultConsumer consumer = new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理
                System.out.println("message receive2: ");
                System.out.println("Received2： " + new String(body, "UTF-8") + ", deliveryTag: " + envelope.getDeliveryTag());
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);// 参数2 false为确认收到消息， true为拒绝收到消息
            }
        };
        channel.basicConsume(queueName,false,consumer);// 参数2 手动确认，代表我们收到消息后需要手动确认告诉服务器我们收到消息了
    }
}
3、Publish/Subscribe

一条消息可以被多个消费者同时获取，生产者将消息发送给交换机，消费者将自己对应的队列注册到交换机，当发送消息后，所有注册的队列的消费者都可以收到消息。

_

发送端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Sender {

    private static String Exchange_Name = "ExchangeDemo";//声明交换机
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        // 声明exchange
        // Producer 将消息发送到 Exchange ，由 Exchange 将消息路由到一个或多个 Queue 中（或者丢弃），Exchange 按照相应的 Binding 逻辑将消息路由到 Queue。
        channel.exchangeDeclare(Exchange_Name,"fanout");
        String message = "Exchange message demo";

        // 消息发送端交换机，如果此时没有队列绑定，则消息会丢失，因为交换机没有存储消息的能力
        channel.basicPublish(Exchange_Name,"",null,message.getBytes());
        System.out.println("发送消息： " + message);
    }
}

消费端示例代码1

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Sub1 {
    private static String Exchange_Name = "ExchangeDemo";//声明交换机
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare("testqueue1",false,false,false,null);
        // 绑定到交换机
        channel.queueBind("testqueue1",Exchange_Name,"");
        channel.basicQos(1);

        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException{

                System.out.println("sub1: " + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("testqueue1",false,consumer);
    }
}

消费端示例代码2

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Sub2 {
    private static String Exchange_Name = "ExchangeDemo";//声明交换机

    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare("testqueue2",false,false,false,null);
        // 绑定到交换机
        channel.queueBind("testqueue2",Exchange_Name,"");
        channel.basicQos(1);

        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException{
                System.out.println("sub2: " + new String(body));
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("testqueue2",false,consumer);
    }
}
4、Routing

生产者将消息发送到type为direct模式的交换机，消费者的队列将自己绑定到路由的时候给自己绑定一个key，只有生产者发送的消息key和绑定的key一致时，消费者才能收到对应的消息。

_

发送端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Sender {
    private static final String ExchangeName = "Rout_Change";//路由消息交换机

    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(ExchangeName,"direct");
        channel.basicPublish(ExchangeName,"key3",null,"route 消息".getBytes());

        channel.close();
        connection.close();
    }
}

消费端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Sub {
    private static final String ExchangeName = "Rout_Change";//路由消息交换机
    public static void main(String[] args) throws Exception{

        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("testroutequeue1",false,false,false,null);

        // 绑定交换机
        // 参数3 标记 绑定到交换机的时候会有一个标记，只有和它一样标记的消息才会别消费到
        channel.queueBind("testroutequeue1",ExchangeName,"key1");
        channel.queueBind("testroutequeue1",ExchangeName,"key2");//绑定多个标记
        channel.basicQos(1);

        DefaultConsumer consumer = new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理
                System.out.println("message route receive1: ");
                System.out.println("Received1： " + new String(body, "UTF-8") + ", deliveryTag: " + envelope.getDeliveryTag());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);// 参数2 false为确认收到消息， true为拒绝收到消息
            }
        };
        channel.basicConsume("testroutequeue1",false,consumer);// 参数2 手动确认，代表我们收到消息后需要手动确认告诉服务器我们收到消息了
    }
}
5、Topics

该类型与 Direct 类型相似，只是规则没有那么严格，可以模糊匹配和多条件匹配，即该类型 Exchange 使用 Routing key 模式匹配和字符串比较的方式将消息路由至绑定的 Queue。

示例：

Routing key 为 use.stock 的消息会转发给绑定匹配模式为 .stock, use.stock， . 和 #.use.stock.# 的 Queue； 表是匹配一个任意词组，# 表示匹配 0 个或多个词组。

_

发送端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Sender {
    private static String Exchange_Name = "Exchange_Topic";
    public static void main(String[] args) throws Exception{

        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        // 声明exchange类型为Topic 也就是通配符模式
        channel.exchangeDeclare(Exchange_Name,"topic");
        channel.basicPublish(Exchange_Name,"abc.1.2",null,"Topic 模式消息".getBytes());

        // 关闭通道和连接
        channel.close();
        connection.close();
    }
}

接收端示例代码

import AMQP.RabbitMQTutorials.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Sub {
    private static String ExchangeName = "Exchange_Topic";
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("topicqueue",false,false,false,null);
        // 绑定交换机
        // 参数3 标记 绑定到交换机的时候会有一个标记，只有和它一样标记的消息才会别消费到
        channel.queueBind("topicqueue",ExchangeName,"key.*");
        channel.queueBind("topicqueue",ExchangeName,"abc.#");//绑定多个标记
        channel.basicQos(1);

        DefaultConsumer consumer = new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                //接收到的消息，进行业务逻辑处理
                System.out.println("message route receive1: ");
                System.out.println("Received1： " + new String(body, "UTF-8") + ", deliveryTag: " + envelope.getDeliveryTag());
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(), false);// 参数2 false为确认收到消息， true为拒绝收到消息
            }
        };
        channel.basicConsume("topicqueue",false,consumer);// 参数2 手动确认，代表我们收到消息后需要手动确认告诉服务器我们收到消息了
    }
}

参考链接

RabbitMQ Tutorials

Rabbitmq企业级消息队列视频课程

文章标签：
云消息队列 MQ
消息中间件
Java
JavaScript
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
taro_秋刀鱼
目录
相关文章
hhzz
|
1月前
|
消息中间件 存储 Serverless
【实践】快速学会使用阿里云消息队列RabbitMQ版
云消息队列 RabbitMQ 版是一款基于高可用分布式存储架构实现的 AMQP 0-9-1协议的消息产品。云消息队列 RabbitMQ 版兼容开源 RabbitMQ 客户端，解决开源各种稳定性痛点（例如消息堆积、脑裂等问题），同时具备高并发、分布式、灵活扩缩容等云消息服务优势。
hhzz
84 2
Yaiba123
|
3月前
|
消息中间件 网络架构
RabbitMQ消息队列常见面试题
这篇文章总结了RabbitMQ的常见面试题，涵盖了消息模型、使用场景、实现功能、消息幂等性、顺序性、堆积和丢失的避免方法，以及推模式和拉模式的区别。
Yaiba123
68 0
开朗觉觉
|
6月前
|
消息中间件 JSON Java
RabbitMQ消息队列
RabbitMQ消息队列
开朗觉觉
64 0
听白
|
6月前
|
消息中间件 数据采集 Serverless
消息队列RabbitMQ版
消息队列RabbitMQ版
听白
42 0
好学的小z
|
6月前
|
消息中间件 存储 负载均衡
消息队列学习之RabbitMQ
【4月更文挑战第3天】消息队列学习之RabbitMQ，一种基于erlang语言开发的流行的开源消息中间件。
好学的小z
58 0
1841928029236104
|
消息中间件 存储 网络协议
RabbitMQ消息队列常见面试题总结
RabbitMQ消息队列常见面试题总结
1841928029236104
573 0
RabbitMQ消息队列常见面试题总结
沐雨风栉
|
消息中间件 存储 自然语言处理
RabbitMQ：深入理解消息队列
前言 在分布式系统中，消息队列是一种常用的解耦和通信方式。它允许应用程序之间通过消息的形式进行通信，从而降低了系统各部分之间的耦合。RabbitMQ是一款开源的、高性能的、可扩展的消息队列系统，它基于AMQP协议实现。本篇博客将带你深入理解RabbitMQ的原理和实践，让你更好地运用这一技术。
沐雨风栉
339 0
北溪入江流
|
消息中间件 NoSQL Java
消息队列、RabbitMQ和RocketMQ
消息队列、RabbitMQ和RocketMQ
北溪入江流
130 0
云课程笔记
|
消息中间件 存储 新零售
初步了解消息队列 RabbitMQ 版｜学习笔记
快速学习初步了解消息队列 RabbitMQ 版
云课程笔记
172 0
pgnozxzkp4mgq
|
消息中间件 Kafka RocketMQ
【消息队列系列6】RabbitMQ使用姿势
主要讲述RabbitMQ常用的使用姿势。
pgnozxzkp4mgq
206 0
【消息队列系列6】RabbitMQ使用姿势
云原生

云消息队列

热门文章

最新文章

  • 1
    Redis Stream——作为消息队列的典型应用场景
  • 2
    消息队列（MQ）重磅推出MQTT移动物联套件
  • 3
    ActiveMQ消息队列
  • 4
    Redis笔记（七）Java实现Redis消息队列
  • 5
    滴滴出行基于RocketMQ构建企业级消息队列服务的实践
  • 6
    RabbitMQ消息队列（九）：Publisher的消息确认机制
  • 7
    Kafka是什么，JMS是什么，常见的类JMS消息服务器，为什么需要消息队列（来自学习笔记）
  • 8
    IM开发基础知识补课(五)：通俗易懂，正确理解并用好MQ消息队列
  • 9
    ENode 1.0 - 消息队列的设计思路
  • 10
    RocketMq消息队列使用
  • 1
    云消息队列 RabbitMQ 版实践解决方案评测
    70
  • 2
    RocketMQ体验测评
    64
  • 3
    关键成果与技术应用
    72
  • 4
    消息队列 MQ使用问题之如何使用DefaultMQPushConsumer来消费消息
    154
  • 5
    消息队列 MQ使用问题之遇到消费速度是固定的并且导致了堆积，该怎么办
    180
  • 6
    消息队列 MQ使用问题之如何在grpc客户端中设置负载均衡器
    48
  • 7
    消息队列 MQ使用问题之如何配置生产环境
    108
  • 8
    消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump，该怎么办
    126
  • 9
    消息队列 MQ使用问题之如何将RocketMQ中某个集群的topic迁移到另一个集群
    90
  • 10
    消息队列 MQ操作报错合集之建立连接时发生了超时错误，该如何解决
    506

    • 相关产品

  • 云消息队列 MQ
    文档详情 产品详情

    • 相关课程

    更多
  • 消息队列RabbitMQ入门课程
  • 带您快速了解阿里云消息队列RocketMQ 5.0版
  • 消息队列 RocketMQ 消息集成
  • 消息队列Kafka入门课程
  • 消息队列 RocketMQ 全类型业务消息学习课程
  • 消息队列 MNS 入门课程

    • 相关电子书

    更多
  • 消息队列kafka介绍
  • 云原生消息队列Apache RocketMQ
  • 消息队列 Kafka 版差异化特性

    • 相关实验场景

    更多
  • 快速体验云消息队列RocketMQ版
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 快速入门轻量消息队列（原MNS）
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
    • 下一篇
    阿里云OSS设置跨域访问