Win10下安装RabbitMQ以及基本知识学习

一、为什么选择RabbitMQ?

先说一下场景，这是我们公司遇到，当然我这里不做业务评价哈？虽然我知道他很不合理，但是我是无能为力的。APP端部分注册是Java开发的系统，然后业务端是C#开发的Web程序，具体系统的名字我就不说，万一看到就不好了，需求是这样的当APP注册以后，Web需要对手机号进行信息查找如果是该系统用户那么就将信息统一录入，查看某些统计这样的一个场景吧。当然一开始通过的定时执行任务，这个我不做评价，反正通过一序列办法吧，领导同意使用以及尝试使用消息队列喽，总之梦想达成了吧。接下来就是要使用了，有那么多的MQ中间件，为什么我会选择RabbitMQ，首先我们不需要做太复杂分布式交互，只要能保证消息的稳定和可靠就可以，在那么多的中间件中最稳定，最成熟的就是RabbitMQ，性能也比较好。另外RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持很多乱七八糟的语言，反正基本上基万千宠爱与一身吧，那么我就用它喽，接下来我就说一下心酸的安装经历，悲惨就是看博客开始，于是我决心自己写一篇，把自己的坑给你们说说。

二、安装

电脑是Win10系统，首先在本地安装好环境，然后才能搞起来，接下来就是被博客带到坑里了，大兄弟们真心希望写的详细一些我们并没有那么强大，反正及时刹车，我看了一下官方文档，一切终于搞定，然后就想写一篇以解心中所有不畅快，同时也让大家少走弯路。

https://www.rabbitmq.com/install-windows-manual.html#install-erlang 这个是官方的地址，大家看不懂我的可以去浏览下。

接下来我们这里强调几点东西，然后在开始：

1.安装Erlang环境，基于这个开发所以必须安装；

2.安装RabbitMQ时候会产生2个.erlang.cookie的文件，在Erlang版本从20.2开始，cookie文件的位置是：通常是C： Users 电脑名称 .erlang.cookie，对于RabbitMQ Windows服务：通常是C： WINDOWS system32 config systemprofile；在20.2之前的Erlang版本中，cookie文件的位置和上面位置一样，但是RabbitMQ Windows服务变成C： Windows .erlang.cookie，这里需要做的是需要保证这2个文件内容一种，做法就是将Window下面的文件替换掉RabbitMQ文件下，这一点必须注意，否者就会报下面的错误：

基本上注意到第2点就不会产生什么错误，接下来我们开始安装吧，

Erlang地址：http://erlang.org/download/otp_win64_20.2.exe

RabbitMQ地址：https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.2/rabbitmq-server-3.7.2.exe

下载完成以后首先安装Erlang，很简单一直下一步就好，完成以后配置一下Path的环境变量，如下图所示，右击我的电脑--》管理--》高级系统设置--》环境变量配置--》Path--》编辑--》新建--》你安装的Erlang文件的bin路径；

配置完成以后cmd把小黑框调出来，输入erl，出现下面就完成了；

接下来安装RabbitMQ:也是下一步下一步完成，配置一下环境变量，基本和上面步骤一样但这里不是bin目录，这里是sbin目录，反正环境变量里如下如这样就可以了，

安装完成以后，按照上面注意的第2点操作一下，之后就完成任务了，然后访问下http:localhost:15672,可以访问到说明你成功了;

三、聊聊需要掌握的知识

有心的人会发现这个地方其实和数据库刚开始做连接的时候一样，实现的道理基本是一样的，场景也是一样，生产者--消费者模式，数据库连接池经典的应用场景不能忘，不能忘，我们这个地方不做源码简析，只是对知识的整理。

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue

Queue用于存储消息，类似于阻塞队列的作用；

RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

Exchange(交换机)

生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange，由Exchange将消息路由到一个或多个Queue中。

Routing key

生产者在将消息发送给Exchange的时候会指定一个Routing key，来指定这个消息的路由规则，而这个Routing key需要与Exchange Type及Binding key联合使用才能最终生效。

Binding

通过Binding将Exchange与Queue关联起来，正确的将消息分发到Queue中。

Binding key

在Exchange传递消息给Queue的时候会指定一个Binding key；

在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的Binding key。

Exchange Types

RabbitMQ常用的Exchange Type有四种:Fanout、Direct、Topic、Headers;

Fanout(广播)

Fanout类型的Exchange路由规则非常简单，类似一种广播的机制，

上图中，生产者发送到Exchange的所有消息都会被传递的Queue中。

Direct(发布/订阅)

Direct类型它会把消息由路由传递到Binding key与Routing key完全匹配的Queue中，类似于发布/订阅的模式；

以上图的配置为例，我们以RoutingKey=”error”发送消息到Exchange，则消息会路由到Queue1和Queue2；如果我们以RoutingKey=”info”或RoutingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他RoutingKey发送消息，则消息不会路由到这两个Queue中。

Topic(模糊匹配)

Direct类型的Exchange路由规则是完全匹配Binding key与Routing key。Topic类型的Exchange在匹配规则上进行了扩展，类似与模糊匹配，规则如下：

* ：可以替代一个词；#：可以替代0或者更多的词

Headers

Headers类型是根据发送的消息内容中的Headers属性进行匹配。

在绑定Queue与Exchange时，指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的Headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会被路由传递到Queue，否则不会被传递到该Queue。

参考博客:http://www.importnew.com/25744.html

四、Demo

Queue 一对一模式

public class ProducerQueueDemo {
    public final static String QueueName="rabbitMQ.test";
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        //设置RabbitMQ相关信息
        factory.setHost("localhost");
        //创建一个新的连接
        Connection connection=factory.newConnection();
        //创建一个通道
        Channel channel=connection.createChannel();
        //声明一个队列
        channel.queueDeclare(QueueName,false,false,false,null);
        String message="Hello RabbitMQ";
        //发送消息到队列中
        channel.basicPublish("", QueueName, null, message.getBytes("UTF-8"));
        System.out.println("Producer Send +'" + message + "'");
        //关闭通道和连接
        channel.close();
        connection.close();
    }
}

public class ConsumerQueueDemo {
    private final static String QueueName = "rabbitMQ.test";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.queueDeclare(QueueName, false, false, false, null);
        System.out.println("Customer Waiting Received messages");
        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, "UTF-8");
                System.out.println("Customer Received '" + message + "'");
            }
        };
        //自动回复队列应答 -- RabbitMQ中的消息确认机制
        channel.basicConsume(QueueName, true, consumer);
    }
}

View Code

Queue 一对多模式

public class ProducerQueueDemo2 {
    private static final String QUEUE_NAME="one-to-many";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        for (int i=0;i<10;i++){
            String message="Hello RabbitMQ"+i;
            channel.basicPublish("",QUEUE_NAME,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
            System.out.println("ProducerQueueDemo2 send '"+message+"'");
        }
        channel.close();
        connection.close();
    }
}

public class ConsumerQueueDemo2 {
    private static final String QUEUE_NAME = "one-to-many";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        System.out.print("ConsumerQueueDemo2  Waiting for messages");
        //每次从队列获取的数量
        channel.basicQos(1);
        Consumer consumer=new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("ConsumerQueueDemo2  Received '" + message + "'");
                try {
                    //throw  new Exception();
                    doWork(message);
                }catch (Exception e){
                    channel.abort();
                }finally {
                    System.out.println("ConsumerQueueDemo2 Done");
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        boolean autoAck=false;
        //消息消费完成确认
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String message) {
        try {
            Thread.sleep(1000); // 暂停1秒钟
            System.out.print(message);
        } catch (InterruptedException _ignored) {
            Thread.currentThread().interrupt();
        }
    }
}

public class ConsumerQueueDemo3 {
    private static final String QUEUE_NAME = "one-to-many";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        System.out.print("ConsumerQueueDemo3  Waiting for messages");
        //每次从队列获取的数量
        channel.basicQos(1);
        Consumer consumer=new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("ConsumerQueueDemo3  Received '" + message + "'");
                try {
                    //throw  new Exception();
                    doWork(message);
                }catch (Exception e){
                    channel.abort();
                }finally {
                    System.out.println("ConsumerQueueDemo3 Done");
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        boolean autoAck=false;
        //消息消费完成确认
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String message) {
        try {
            Thread.sleep(1000); // 暂停1秒钟
            System.out.print(message);
        } catch (InterruptedException _ignored) {
            Thread.currentThread().interrupt();
        }
    }
}

View Code

Exchange Fanout(广播)模式

public class ReleaseMessageDemo {
    private static final String EXCHANGE_NAME = "release";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        for (int i=0;i<5;i++){
            String message="这是发布的通知"+i;
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8"));
            System.out.print(message);
        }
        channel.close();
        connection.close();

    }
}

public class SubscribeMessageDemo {
    private static final String EXCHANGE_NAME = "release";
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //随机产生队列
        String queueName=channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println("SubscribeMessageDemo Waiting for messages");
        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, "UTF-8");
                System.out.println("SubscribeMessageDemo Received '" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);//队列会自动删除
    }
}

View Code

Exchange Direct(发布/订阅)模式

public class RoutingDemo {
    private static final String EXCHANGE_NAME = "routing";
    // 路由关键字
    private static final String[] routingKeys = new String[]{"info" ,"warning", "error"};
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        //发送信息
        for (String routingKey:routingKeys){
            String message= "Routing Send the message level:" + routingKey;
            channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));
            System.out.print("Routing Send"+routingKey+":"+message);
        }
        channel.close();
        connection.close();

    }
}

public class RoutingConsumerDemo {
    private static final String EXCHANGE_NAME = "routing";
    // 路由关键字
    private static final String[] routingKeys = new String[]{"info" ,"warning"};
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        //声明交换器
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        String queueName = channel.queueDeclare().getQueue();
        for (String routingKey:routingKeys) {
            channel.queueBind(queueName,EXCHANGE_NAME,routingKey);
            System.out.println("RoutingConsumerDemo exchange:"+EXCHANGE_NAME+"," +
                    " queue:"+queueName+", BindRoutingKey:" + routingKey);
        }
        System.out.println("RoutingConsumerDemo Waiting for messages");
        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, "UTF-8");
                System.out.println("RoutingConsumerDemo Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

public class RoutingConsumerDemo2 {
    private static final String EXCHANGE_NAME = "routing";
    // 路由关键字
    private static final String[] routingKeys = new String[]{"error"};
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        Connection connection=factory.newConnection();
        Channel channel=connection.createChannel();
        //声明交换器
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        String queueName=channel.queueDeclare().getQueue();
        //根据路由关键字进行多重绑定
        for (String severity : routingKeys) {
            channel.queueBind(queueName, EXCHANGE_NAME, severity);
            System.out.println("RoutingConsumerDemo2 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + severity);
        }
        System.out.println("RoutingConsumerDemo2 Waiting for messages");
        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, "UTF-8");
                System.out.println("RoutingConsumerDemo2 Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

View Code

Topics模式这个我就不做demo了，基本如出一辙；Headers模式我没用到过我也不做demo;另外这里需要强调的一点是导入RabbitMQ的jar包；

五、欢迎咨询~

有什么问题加群:438836709,性能方面我没有做很大的压力测，反正能满足我们业务需求吧，准备将好多乱七八糟的系统之间的通信都转成RabbitMQ；