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

2019独角兽企业重金招聘Python工程师标准>>>

我们在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.3 mv -i otp_src_18.3 /usr/local/ //进入/usr/local/otp_src_18.3 cd /usr/local/otp_src_18.3/ //生产安装配置 ./configure --prefix&＃61;/usr/local/erlang

此时报错&＃xff1a;configure: error: no acceptable C compiler found in $PATH

原因是&＃xff1a;缺少gcc的编译环境。

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

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

./configure --prefix&＃61;/usr/local/erlang

再次报错&＃xff1a;

解决&＃xff1a;yum install -y ncurses-devel。ok。

再次执行生产安装配置的操作&＃xff1a;

./configure --prefix&＃61;/usr/local/erlang

还是报错&＃xff1a;

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

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

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

下载unixODBC源码包(http://www.unixodbc.org/unixODBC-2.2.1.tar.gz)放到某处比如/usr/local下&＃xff0c;然后运行下述命令&＃xff1a;

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

报错&＃xff1a;

需要安装flex

yum -y install flex

成功后再次执行

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

成功&＃xff0c;然后执行命令

make && make install

失败&＃xff0c;/bin/sh: yacc: command not found

解决&＃xff1a;yum install -y byacc。ok。

再次执行make && make install&＃xff0c;ok。

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

然后安装openSSL

//在/usr/local mkdir openssl yum list | grep ssl yum install openssl-devel

执行ok&＃xff01;

因为odbc : ODBC library - link check failed报错&＃xff0c;执行以下命令&＃xff1a;

yum list | grep ODBC yum install unixODBC-devel

再次执行erlang的configure命令

./configure --prefix&＃61;/home/erlang --without-javac

执行以下命令&＃xff1a;

yum install gcc-c&＃43;&＃43;

ok&＃xff01;

再次执行./configure --prefix&＃61;/home/erlang --without-javac

崩溃&＃xff01;&＃xff01;&＃xff01;

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

//编译安装 make && make install

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

ok&＃xff01;&＃xff01;&＃xff01;erlang安装ok了。

1.2.配置Erlang的环境变量

vi /etc/profile //在底部加入下面的内容 #set erlang environment ERL_HOME&＃61;/home/erlang PATH&＃61;$ERL_HOME/bin:$PATH export ERL_HOME PATH

让配置生效&＃xff1a;

source /etc/profile //生效

测试

至此&＃xff0c;erlang安装完成&＃xff01;&＃xff01;&＃xff01;

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.xz tar -xvf rabbitmq-server-generic-unix-3.6.1.tar mv -i rabbitmq_server-3.6.1/ /usr/local/

配置rabbitmq环境变量&＃xff1a;

vi /etc/profile //在底部加入以下内容 #set rabbitmq environment export PATH&＃61;$PATH:/usr/local/rabbitmq_server-3.6.1/sbin //让配置生效&＃xff0c;执行以下命令 source /etc/profile

启动服务&＃xff1a;

//启动rabbitmq&＃xff0c;-detached代表后台守护进程方式启动。 rabbitmq-server -detached

查看状态&＃xff1a;

rabbitmqctl status

如果能显示rabbitmq的状态信息&＃xff0c;表明安装成功&＃xff01;

Status of node rabbit&＃64;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.配置管理界面插件

首先创建目录&＃xff0c;否则可能报错&＃xff1a;

mkdir /etc/rabbitmq

然后启用插件&＃xff1a;

mkdir /etc/rabbitmq rabbitmq-plugins enable rabbitmq_management

配置防火墙&＃xff1a;

配置linux 端口。网页管理端口&＃xff1a;15672 AMQP端口&＃xff1a; 5672

firewall-cmd --permanent --add-port&＃61;15672/tcp firewall-cmd --permanent --add-port&＃61;5672/tcp systemctl restart firewalld.service

访问RabbitMQ管理页面&＃xff1a;http://192.168.1.177:15672/

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

新建用户并设置权限

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

使用用户名&＃xff1a;lyang&＃xff0c;密码&＃xff1a;lyang登陆&＃xff0c;ok&＃xff01;

4.Java客户端连接

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

pom.xml中加入依赖&＃xff1a;

org.springframework.bootspring-boot-starter-amqp

application.yml中加入RabbitMQ的连接配置&＃xff1a;

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

此时基本配置已完成&＃xff0c;可以开始码代码了&＃xff1a;

生产者&＃xff1a;RabbitProducer

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

消费者&＃xff1a;RabbitConsumer

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

运行&＃xff01;