RabbitMQ安装以及消息订阅模式

参考1
参考2

1. 安装预环境

安装GCC GCC-C++ Openssl等模块
yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel
安装ncurses
yum -y install ncurses-devel
yum install unixODBC-devel
yum -y install gtk2-devel binutils-devel
yum -y install mesa-libGLU-devel

2. 安装erlang

cd /usr/local/src
wget http://erlang.org/download/otp_src_18.2.1.tar.gz
tar -zxvf otp_src_18.2.1.tar.gz
cd otp_src_18.2.1
./configure --prefix=/usr/local/erlang --without-javac 
make && make install

若有提示如下：

jinterface     : Java compiler disabled by user
odbc           : ODBC library - link check failed
wxWidgets not found, wx will NOT be usable
Using fakefop to generate placeholder PDF files.

则：

• 安装wxWidgets

wget http://prdownloads.sourceforge.net/wxwindows/wxWidgets-2.8.12.tar.gz
tar -zxvf wxWidgets-2.8.12.tar.gz
cd wxWidgets-2.8.12
./configure --with-gtk --enable-unicode --disable-shared
./configure --with-opengl --enable-debug --enable-unicode
make && make install

3. rabbitmq安装

#下载地址
wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.9/rabbitmq-server-generic-unix-3.6.9.tar.xz    
tar -xvf rabbitmq-server-generic-unix-3.6.9.tar
#移动至/usr/local/下 改名rabbitmq
cp -r rabbitmq_server-3.6.9 /usr/local/rabbitmq

4. 配置环境变量
   vi /etc/profile

#erlang
export PATH=$PATH:/usr/local/erlang/bin
#RABBITMQ
export MQ_HOME=/usr/local/rabbitmq
export PATH=$MQ_HOME/sbin:$PATH

source /etc/profile //使修改生效

5. 验证是否安装成功

• 运行如下命令，验证erlang是否安装成功。
  erl -version

Erlang (SMP,ASYNC_THREADS,HIPE) (BEAM) emulator version 7.2.1

6. 开启RabbitMQ的Web监控
   cd /usr/local//rabbitmq
   rabbitmq-plugins enable rabbitmq_management

7. 启动RabbitMQ
   rabbitmq-server –detached
   停止：
   rabbitmqctl stop_app
   或
   service rabbitmq-server start #启动
   service rabbitmq-server stop #停止
   service rabbitmq-server restart #重启
   service rabbitmq-server status #查看状态
   service rabbitmq-server etc #查看有哪些命令可以使用

8. 加入开机自启动服务
   chkconfig rabbitmq-server on

9. 添加用户和权限
   cd /usr/local/rabbitmq/sbin
   默认网页guest用户是不允许访问的，需要增加一个用户修改一下权限，代码如下：
   添加用户:
   ./rabbitmqctl add_user admin admin
   添加权限:
   ./rabbitmqctl set_permissions -p "/" admin "." "." ".*"
   修改用户角色:
   ./rabbitmqctl set_user_tags admin administrator
   然后就可以远程访问了，然后可直接配置用户权限等信息。
   http://129.211.24.177:15672/
   

   访问例子

设置topic MQ

1. 新建用户xies/12345678

   
   image.png

2. 设置vhost
   新建vhost,如下图,右侧选择virtual hosts：

   
   image.png
   

   下拉倒页面底部,输入vhost名称，如hscs_vhost，点击add virtual host

3. 设置queues
   新建queues,如下图,选择queues：

   
   image.png
   

   下拉倒页面底部，
   选择vhost，如hscs_vhost，
   选择queue名name，如ALIX_QUES，
   选择durable，auto delete选择No，点击add queue

4. 设置exchanges
   新建exchanges,如下图,选择exchanges：

   
   image.png
   

   下拉倒页面底部，
   选择vhost，如hscs_vhost，
   选择exchange名name，如HSCS_E，
   选择type,如topic，
   选择durable，auto delete选择No，点击add exchange

5. Exchanges绑定queues
   点击如下图Exchanges的name，如HSCS_E
   输入已有queue的名称，如ALIX_QUES
   输入一个routing key,如ALIX_KEY
   点击绑定bing

   
   image.png

6. 查询queue绑定的routing key值
   点击如下图queues的name，如ALIX_QUES
   查看绑定的routing key

   
   image.png

topic MQ代码展示

默认端口是5672

• pom依赖

        <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.7.1</version>
        </dependency>

1. 生产者代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created by hand on 2018/5/2.
 */
public class Producer {
    private static final String EXCHANGE_NAME = "HSCS_E";
    // 路由关键字
    private static final String[] routingKeys = new String[]{"ALIX_KEY" };
    //持久化
    private static final boolean durable = true;
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("129.211.24.177");
        factory.setPort(5672);
        factory.setUsername("xies");
        factory.setPassword("12345678");
        factory.setVirtualHost("hscs_vhost");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //声明转发器
        channel.exchangeDeclare(EXCHANGE_NAME, "topic",durable);
        //发送信息
        for (String routingKey:routingKeys){
            String message = "测试消息为:" + routingKey;
            //推送消息
            channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes());
//            channel.d
            System.out.println("Producer"+message);
        }
        channel.close();
        connection.close();
    }

2. 消费者代码

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class AlixConsumer {
    private static String  QUEUE="ALIX_QUES";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("129.211.24.177");
        factory.setPort(5672);
        factory.setUsername("xies");
        factory.setPassword("12345678");
        factory.setVirtualHost("hscs_vhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        System.out.println("Consumer  Waiting for messages");
        com.rabbitmq.client.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("Consumer接收到'" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(QUEUE, true, consumer);
    }