RabbitMQ---安装、运行、java客户端连接-白红宇

RabbitMQ---安装、运行、java客户端连接

阅读量：6853 次

发布时间：2019-06-26

本文共 9884 字，大约阅读时间需要 32 分钟。

我们在centos7上安装RabbitMQ。

1.安装Erlang

1.1.下载安装

//下载安装包文件wget http://www.erlang.org/download/otp_src_18.3.tar.gz//解压缩tar -zvxf otp_src_18.3.tar.gz//移动解压缩后文件夹。/usr/local/src/otp_src_18.3    --> /usr/local/otp_src_18.3mv -i  otp_src_18.3   /usr/local///进入/usr/local/otp_src_18.3cd /usr/local/otp_src_18.3///生产安装配置./configure --prefix=/usr/local/erlang

此时报错：configure: error: no acceptable C compiler found in $PATH

原因是：缺少gcc的编译环境。

执行yum install -y gcc安装gcc编译环境。如果yum可以先安装yum，或者直接gcc的安装包。安装ok

再次执行生产安装配置的操作

./configure --prefix=/usr/local/erlang

再次报错：

解决：yum install -y ncurses-devel。ok。

再次执行生产安装配置的操作：

./configure --prefix=/usr/local/erlang

还是报错：

因为我们用的gcc编译，所以不需要java compiler,这个提示不用管。

---------------------------------不关注这个------------------------------------------------

需要关注odbc : ODBC library – link check failed 表示你未安装unixODBC库

下载unixODBC源码包()放到某处比如/usr/local下，然后运行下述命令：

//在/usr/local下mkdir  unixODBC-2.2.1cd unixODBC-2.2.1./configure --prefix=/usr/local/unixODBC-2.2.1 --includedir=/usr/include --libdir=/usr/lib -bindir=/usr/bin --sysconfdir=/etc --enable-gui=no

报错：

需要安装flex

yum -y install flex

成功后再次执行

./configure --prefix=/usr/local/unixODBC-2.2.1 --includedir=/usr/include --libdir=/usr/lib -bindir=/usr/bin --sysconfdir=/etc --enable-gui=no

成功，然后执行命令

make && make install

失败，/bin/sh: yacc: command not found

解决：yum install -y byacc。ok。

再次执行make && make install，ok。

-------------------------------------------end---------------------------------------------------------------

然后安装openSSL

//在/usr/localmkdir opensslyum list | grep sslyum install openssl-devel

执行ok！

因为odbc : ODBC library - link check failed报错，执行以下命令：

yum list | grep ODBC  yum install unixODBC-devel

再次执行erlang的configure命令

./configure --prefix=/home/erlang --without-javac

执行以下命令：

yum install gcc-c++

ok！

再次执行./configure --prefix=/home/erlang --without-javac

崩溃！！！

哈哈，其实这个错不用管了，我们用的gcc编译，直接安装。

//编译安装make && make install

测试erlang是否安装成功。/home/erlang/bin/erl。

ok！！！erlang安装ok了。

1.2.配置Erlang的环境变量

vi /etc/profile//在底部加入下面的内容#set erlang environmentERL_HOME=/home/erlangPATH=$ERL_HOME/bin:$PATHexport ERL_HOME PATH

让配置生效：

source /etc/profile  //生效

测试

至此，erlang安装完成！！！

2.安装RabbitMQ

2.1.下载安装

wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-generic-unix-3.6.1.tar.xzxz -d rabbitmq-server-generic-unix-3.6.1.tar.xztar -xvf rabbitmq-server-generic-unix-3.6.1.tarmv -i rabbitmq_server-3.6.1/ /usr/local/

配置rabbitmq环境变量：

vi /etc/profile//在底部加入以下内容#set rabbitmq environmentexport PATH=$PATH:/usr/local/rabbitmq_server-3.6.1/sbin//让配置生效，执行以下命令source /etc/profile

启动服务：

//启动rabbitmq，-detached代表后台守护进程方式启动。rabbitmq-server -detached

查看状态：

rabbitmqctl status

如果能显示rabbitmq的状态信息，表明安装成功！

Status of node rabbit@localhost ...[{pid,37234}, {running_applications,[{rabbit,"RabbitMQ","3.6.1"},                        {mnesia,"MNESIA  CXC 138 12","4.13.3"},                        {rabbit_common,[],"3.6.1"},                        {ranch,"Socket acceptor pool for TCP protocols.",                               "1.2.1"},                        {os_mon,"CPO  CXC 138 46","2.4"},                        {xmerl,"XML parser","1.3.10"},                        {sasl,"SASL  CXC 138 11","2.7"},                        {stdlib,"ERTS  CXC 138 10","2.8"},                        {kernel,"ERTS  CXC 138 10","4.2"}]}, {os,{unix,linux}}, {erlang_version,"Erlang/OTP 18 [erts-7.3] [source] [64-bit] [smp:2:2] [async-threads:64] [hipe] [kernel-poll:true]\n"}, {memory,[{total,42598496},          {connection_readers,0},          {connection_writers,0},          {connection_channels,0},          {connection_other,0},          {queue_procs,2808},          {queue_slave_procs,0},          {plugins,0},          {other_proc,18822224},          {mnesia,60888},          {mgmt_db,0},          {msg_index,39096},          {other_ets,900800},          {binary,27024},          {code,17392316},          {atom,662409},          {other_system,4690931}]}, {alarms,[]}, {listeners,[{clustering,25672,"::"},{amqp,5672,"::"}]}, {vm_memory_high_watermark,0.4}, {vm_memory_limit,764742860}, {disk_free_limit,50000000}, {disk_free,10780426240}, {file_descriptors,[{total_limit,924},                    {total_used,2},                    {sockets_limit,829},                    {sockets_used,0}]}, {processes,[{limit,1048576},{used,138}]}, {run_queue,0}, {uptime,19}, {kernel,{net_ticktime,60}}]

3.配置管理界面插件

首先创建目录，否则可能报错：

mkdir /etc/rabbitmq

然后启用插件：

mkdir /etc/rabbitmqrabbitmq-plugins enable rabbitmq_management

配置防火墙：

配置linux 端口。网页管理端口：15672 AMQP端口： 5672

firewall-cmd --permanent --add-port=15672/tcpfirewall-cmd --permanent --add-port=5672/tcpsystemctl restart firewalld.service

访问RabbitMQ管理页面：http://192.168.1.177:15672/

使用用户名：guest，密码：guest登陆，发现登陆不了，由于账号guest具有所有的操作权限，并且又是默认账号，出于安全因素的考虑，guest用户只能通过localhost登陆使用，并建议修改guest用户的密码以及新建其他账号管理使用rabbitmq(该功能是在3.3.0版本引入的)。

新建用户并设置权限

rabbitmqctl add_user lyang lyang  //添加用户，后面两个参数分别是用户名和密码，我这都用lyang了。rabbitmqctl set_permissions -p / lyang ".*" ".*" ".*"  //添加权限rabbitmqctl set_user_tags lyang administrator  //修改用户角色

使用用户名：lyang，密码：lyang登陆，ok！

4.Java客户端连接

这里我们用SpringBoot的方式连接。

pom.xml中加入依赖：


        
     
      org.springframework.boot
         
     
      spring-boot-starter-amqp

application.yml中加入RabbitMQ的连接配置：

spring:  rabbitmq:    host: 192.168.1.177    port: 5672    username: guest    password: guest

此时基本配置已完成，可以开始码代码了：

生产者：RabbitProducer

import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeUnit;public class RabbitProducer {    private static final String EXCHANGE_NAME = "exchange_demo";    private static final String ROUTING_KEY = "routingkey-demo";    private static final String QUEUE_NAME = "queue_demo";    private static final String IP_ADDRESS = "192.168.1.177";    private static final int PORT = 5672;      //rabbitmq默认端口号5672    private static final String USERNAME = "lyang";    private static final String PASSWORD = "lyang";    public static void main(String[] args) throws Exception {        //创建rabbitmq的ConnectionFactory（连接工厂）        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(IP_ADDRESS);        connectionFactory.setPort(PORT);        connectionFactory.setUsername(USERNAME);        connectionFactory.setPassword(PASSWORD);        //创建连接        Connection connection = connectionFactory.newConnection();        //创建通道        Channel channel = connection.createChannel();        //创建一个交换器        //type=direct、持久化的、非自动删除的交换器        //第三个参数boolean durable设置是否持久化，持久化可以将交换器存盘，服务器重启时不会丢失相关信息        //第四个参数boolean autoDelete设置是否自动删除，        channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT,true,false,null);        //将交换器与队列通过路由键绑定        channel.queueDeclare(QUEUE_NAME,true,false,false,null);        //将交换器和队列通过路由键绑定        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);        //发送一条持久化的消息        String msg = "hello world!";        //发布消息        //第三个参数是boolean mandatory,表示当无法根据路由键找到一个符合条件的队列时，        //rabbitmq会调用Basic.Return命令将消息返回给生产者        //第四个参数是boolean immediate,表示队列没有任何消费者时，这条消息会被返回给生产者        channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,true,false,                    MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());        //添加ReturnListener来获取没有被正确路由到合适队列的消息        channel.addReturnListener(new ReturnListener() {            @Override            public void handleReturn(int replyCode, String replyText,                      String exchange, String rountingKey,AMQP.BasicProperties basicProperties,                       byte[] bytes) throws IOException {                String msg = new String(bytes);                System.out.println("Basic.return返回的结果是："+msg);            }        });        TimeUnit.SECONDS.sleep(200);        channel.close();        connection.close();    }}

消费者：RabbitConsumer

import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeUnit;public class RabbitConsumer {    private static final String QUEUE_NAME = "queue_demo";    private static final String IP_ADDRESS = "192.168.1.177";    private static final int PORT = 5672;      //rabbitmq默认端口号5672    private static final String USERNAME = "guest";    private static final String PASSWORD = "guest";    public static void main(String[] args) throws Exception{        //创建地址        Address[] addresses = new Address[]{new Address(IP_ADDRESS,PORT)};        //创建ConnectionFactory        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setUsername(USERNAME);        connectionFactory.setPassword(PASSWORD);        //创建连接        Connection connection = connectionFactory.newConnection(addresses);        //创建通道        final Channel channel = connection.createChannel();        //设置客户端最多接收未被ack的消息的个数        channel.basicQos(64);        //创建消费端        Consumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope,                  AMQP.BasicProperties properties, byte[] body) throws IOException{                System.out.println("receive msg = "+new String(body));                try{                    TimeUnit.SECONDS.sleep(1);                }catch(InterruptedException e){                    e.printStackTrace();                }                //deliveryTag可以看做成消息的编号，64位的长整型                //第二个参数requeue,表示是否将消息重新存入队列，以便可以发送给下一个消费者                channel.basicAck(envelope.getDeliveryTag(),true);  //显示确认接收消息                //channel.basicReject(envelope.getDeliveryTag(),true);  //显示拒绝接受消息            }        };        //第二个参数false表示取消自动确认消息接收机制、改为显示手动确认，防止消息丢失        channel.basicConsume(QUEUE_NAME,false,consumer);           //等待回调函数执行完毕之后，关闭资源        TimeUnit.SECONDS.sleep(200);        channel.close();        connection.close();    }}

运行！