RabbitMQ安装以及消息订阅模式

安装预环境

安装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

安装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

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

配置环境变量
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 //使修改生效

验证是否安装成功

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

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

开启RabbitMQ的Web监控
cd /usr/local//rabbitmq
rabbitmq-plugins enable rabbitmq_management
启动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 #查看有哪些命令可以使用
加入开机自启动服务
chkconfig rabbitmq-server on
添加用户和权限
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

新建用户xies/12345678

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

image.png

下拉倒页面底部,输入vhost名称，如hscs_vhost，点击add virtual host
设置queues
新建queues,如下图,选择queues：

image.png

下拉倒页面底部，
选择vhost，如hscs_vhost，
选择queue名name，如ALIX_QUES，
选择durable，auto delete选择No，点击add queue
设置exchanges
新建exchanges,如下图,选择exchanges：

image.png

下拉倒页面底部，
选择vhost，如hscs_vhost，
选择exchange名name，如HSCS_E，
选择type,如topic，
选择durable，auto delete选择No，点击add exchange
Exchanges绑定queues
点击如下图Exchanges的name，如HSCS_E
输入已有queue的名称，如ALIX_QUES
输入一个routing key,如ALIX_KEY
点击绑定bing

image.png
查询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>

生产者代码

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();
    }

消费者代码

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);
    }