团长大人的学习笔记——RabbitMQ

时间  2019-11-08
标签 团长 大人 学习 笔记 rabbitmq 栏目 RabbitMQ 繁體版
原文   原文链接

1、RibbitMQ的基础介绍

1. 为何要使用MQ 2. 与其余MQ的区别java

  • **ActiveMQ：**使用Java开发，遵循JMS规范，使用方便，支持多种协议。可是有丢失消息的风险而且速度较慢
  • **RabbitMQ：**使用Erlang开发（用于解决高并发的问题），能够解决并发问题。可是只支持AMQP协议且不能动态扩展

2、RabbitMQ的安装

1. 安装Erlang环境(这一步参照博客 www.jianshu.com/p/27197d58e…)c++

  • 安装阿里的yum源（我在安装的时候下载速度很慢，因此这边使用阿里的yum源来安装erlang）算法

    wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-6.repo
yum-y install make gcc gcc-c++kernel-devel m4 ncurses-devel openssl-devel java-devel  unixODBC-devel
复制代码

  • 安装erlang的yum源shell

    wget http://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm
rpm -Uvh erlang-solutions-1.0-1.noarch.rpm
rpm --import http://packages.erlang-solutions.com/rpm/erlang_solutions.asc
复制代码

  • 而后就能够直接安装erlang了bash

    yum -y install erlang
复制代码

  • 经过下面的命令就能够查看是否安装完成了服务器

    erl
复制代码

2. 安装RibbitMQ数据结构

  • 下载Rabbit的yum源（下载下来的时候名字很乱，改个名字）并发

    wget https://bintray.com/rabbitmq/rpm/download_file?file_path=rabbitmq-server%2Fv3.7.x%2Fel%2F7%2Fnoarch%2Frabbitmq-server-3.7.14-1.el7.noarch.rpm
mv download_file\?file_path\=rabbitmq-server%2Fv3.7.x%2Fel%2F7%2Fnoarch%2Frabbitmq-server-3.7.14-1.el7.noarch.rpm rabbitmq-server-3.7.14-el7.noarch.rpm
复制代码

  • 安装app

    yum -y install rabbitmq-server-3.7.14-el7.noarch.rpm
复制代码
    • 若是报有依赖须要解决，就直接使用yum下载这个依赖就行了

  • 启动服务ide

    rabbitmq-server start
复制代码

  • 后台启动

    rabbitmq-server -datached
复制代码

  • 启动后使用amqp协议，默认在5672端口

3、RabbitMQ初步使用

1. 搭建管理平台

  • 初步搭建没有任何插件，咱们使用下面的命令下载并启用RabbitMQ的管理地址

    rabbitmq-plugins enable rabbitmq_management
复制代码

  • 如今就能够访问该节点的15672端口使用guest/guest来登录管理界面

    • 若是不是在localhost下访问，咱们还须要修改/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.14/ebin/rabbit.app文件，将{loopback_users, [<<”guest”>>]} 改成{loopback_users, []}

      {default_user, <<"guest">>},
{default_pass, <<"guest">>},
{default_user_tags, [administrator]},
{default_vhost, <<"/">>},
{default_permissions, [<<".*">>, <<".*">>, <<".*">>]},
{loopback_users, []},
{password_hashing_module, rabbit_password_hashing_sha256},
{server_properties, []},
复制代码

      再重启就OK了

    • Virtural Host 用于区分不一样业务，每一个VH都是独立的，互不影响的。不一样的团队用不一样的VH，相互隔离

2. 点对点简单队列

点对点简单队列：一个生产者投递消息给队列，只容许一个消费者进行消费，（若是存在消费者集群，则会均摊消费，使用取模算法）每一个消息只会消费一次

生产者生产的消息直接投递给队列服务器，而后队列服务器直接推送或消费者自行拉取消息

  • ACK应答模式

    • 自动应答：当消费者收到消息后，不管是否处理，消费者都会自动应答消费。
    • 手动应答：消费者在代码里显式的回复ACK

  • 导入依赖

    <dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.0.2</version>
</dependency>
复制代码

  • 封装一个连接工具

    public class MQConnectionUtils {
    /** * 建立新的连接 * @return */
    public static Connection connect() throws IOException, TimeoutException {
        //建立连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();

        //设置连接参数
        connectionFactory.setHost("192.168.3.203");
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        return connectionFactory.newConnection();
    }
}
复制代码

  • 生产者

    public class Producer {
    /**队列名称*/
    private static final String QUEUE_NAME = "libi_QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = MQConnectionUtils.connect();
        //建立通道
        Channel channel = connection.createChannel();
        //建立一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //建立消息
        String message = "Libi_Message";
        //发送消息
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        //关闭通道和连接
        channel.close();
        connection.close();
        System.out.println("消息投递成功！");
    }
}
复制代码

  • 消费者

    public class Consumer {
    /**队列名称*/
    private static final String QUEUE_NAME = "libi_QUEUE";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接
        Connection connection = MQConnectionUtils.connect();
        //建立通道
        Channel channel = connection.createChannel();
        //消费者关联一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            //使用匿名内部类重写获取消息的方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body, "UTF-8");
                System.out.println("活动生产者消息："+msg);
            }
        };
        //设置应答模式，true表示自动应答，false表示手动应答
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
        //关闭通道和连接
        channel.close();
        connection.close();
        System.out.println("消息消费成功！");
    }
}
复制代码

3. 公平队列（须要本身在代码里实现）

  • 均摊消费的缺点：当消费者处理消息的能力不一致时，若是仍是均摊处理信息，则会形成资源浪费（对消费慢点节点不公平），须要实现"能者多劳"

  • 公平队列的实现思路（BaseQos方法）：当有n个消费者在上一条消息尚未处理完成时（尚未发送ACK），消息队列就不会发送下一条消息给它，给另一个消费者。在生产者经过以下代码开启Qos

    channel.basicQos(n);
复制代码

  • 若是这时消费者在代码里忘记应答了，那么就会陷入阻塞

4.发布订阅模式

生产者投递消息给交换机，交换机根据路由策略转发到不一样的队列服务器中，队列服务器再给消费者进行消费

  • 交换机策略

    • Direct：直接交换机，一种带路由功能的交换机，一个队列会和一个交换机绑定，除此以外再绑定一个routing_key，当消息被发送的时候，须要指定一个binding_key，这个消息被送达交换机的时候，就会被这个交换机送到指定的队列里面去。一样的一个binding_key也是支持应用到多个队列中的。

      就是说直接交换机能够更具生产者的routing_key和消费者的binding_key进行匹配，只有同样才会转发这个消息

    • Fanout：扇形交换机，它所能作的事情很是简单———广播消息。扇形交换机会把能接收到的消息所有发送给绑定在本身身上的队列。由于广播不须要“思考”，因此扇形交换机处理消息的速度也是全部的交换机类型里面最快的。

    • Topic：主题交换机，发送到主题交换机上的消息须要携带指定规则的routing_key，主题交换机会根据这个规则将数据发送到对应的(多个)队列上。

      主题交换机的routing_key须要有必定的规则，交换机和队列的binding_key须要采用*.#.*.....的格式，每一个部分用.分开，其中：

      • *表示一个单词
      • #表示任意数量（零个或多个）单词。

      假设有一条消息的routing_key为fast.rabbit.white,那么带有这样binding_key的几个队列都会接收这条消息：

    • Handler：首都交换机，首部交换机是忽略routing_key的一种路由方式。路由器和交换机路由的规则是经过Headers信息来交换的，这个有点像HTTP的Headers。将一个交换机声明成首部交换机，绑定一个队列的时候，定义一个Hash的数据结构，消息发送的时候，会携带一组hash数据结构的信息，当Hash的内容匹配上的时候，消息就会被写入队列。

  • 生产者（在发送消息的时候传入exchange的参数）

    /** * 使用Fanout类型的交换机,交换器转给发所有的队列 */
public class Producer {
    //交换机名称
    static final String EXCHANGE_NAME = "fanout_destination";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQConnectionUtils.connect();
        Channel channel = connection.createChannel();

        //绑定交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        String msg = "my_fanout_meg";
        //发送消息(路由策略为空串)
        channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());

        channel.close();
        connection.close();
    }
}
复制代码

  • 消费者

    public class EmailConsumer {
    private static String QUEUE_NAME = "Email_Queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = MQConnectionUtils.connect();
        Channel channel = connection.createChannel();

        //消费者声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //消费者绑定交换机
        channel.queueBind(QUEUE_NAME, Producer.EXCHANGE_NAME,"");
        //监听消息
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println("邮件消费者：" + msg);
            }
        };

        channel.basicConsume(QUEUE_NAME, defaultConsumer);
    }
}
复制代码

    消费者的chanel和connection没有关闭，能够多启动几个，就会发现全部的消费者均可以收到生产者传入的信息

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程