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

RabbitMq的基本认识和配置

什么是消息中间件

消息：在应用之间传递的数据
消息中间件：Message Queue Middleware，简称MQ，是指利用高效可靠的消息机制进行与平台无关的数据交流，并基于数通信机制来进行分布式系统的集成。通过提供的消息传递和消息排队模型，它可以在分布式环境下拓张进程间的通信.
传递的模式： 1.点对点(P2P,Point-to-Point)

2.发布订阅模式(Pub/Sub)

消息中间件适用的方向: 适用于可靠的数据传输的分布式环境。能够在不同的平台下通信，屏蔽各个平台以及协议之间的特性，实现应用程序的链接，并且在任何时刻可以将消息进行传送或者存储转发，这比远程过程调用更加进步。

消息中间件的作用

解耦：在项目的启动之初来预测将来会碰到什么需求是极其困难的。消息中间件在处理过程中间插入一个隐含的，基于数据的接口层，两边的处理过程都要实现这一个接口，这允许你独立的拓展或修改两边的处理过程，只要确保他们遵循同样的接口约束即可。
冗余(存储)：有些情况下，处理数据的过程会失败。消息中间件可以把数据进行持久化直到他们已经被完全处理，通过这一方式规避了数据丢失的风险。在把一个消息从消息中间件中删除之前，需要你的处理系统明确地指出这个消息被处理完成，从而确保你地数据被安全地保存知道你使用完毕。
扩展性：消息中间件解耦了应用地处理过程，所以提供了消息入队和处理地效率是很容易地，只需要另外增加处理过程即可，不需要改变代码，也不需要调节参数。
削峰：在访问量剧增地情况下，应用仍然需要继续发挥作用，但是这个月地突发流量的情况的不常见。
可恢复性：当系统一部分的组件失效时，不会影响到整个系统。消息中间件降低了进程间的耦合度。
顺序保证：在大所属使用场景下，数据处理顺序很重要，大部分消息中间件支持一定程度上的顺序性。
缓冲:在任何重要的系统中，都会存在需要不同处理时间的元素。消息中间件通过一个缓冲层来帮助任务最高效率的执行，写入消息中间件的处理会尽可能快速。
异步通信：在很多时候应用不想也不需要立即处理消息。消息中间件提供了异步通信机制，允许应用把一些消息放入消息中间件中，但并不立即处理它，在之后需要的时候再慢慢处理。

Rabbitmq安装

安装erlang

因为RabbitMQ采用Erlang语言编写的，所以需要配置erlang的环境,配置如下:

1.安装gcc和openssl模块

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

2.安装

yum -y install ncurses-devel

3.指定编译后程序的路径

./configure --prefix=/usr/erlang

4.编译安装

make

make install

5.配置环境变量

配置好之后使用source /etc/profile将该配置文件生效,生效之后输入erl回车可以看到：

安装完成erlang的环境之后，需要安装的就是rabiitmq了。

1.这里我下载的时rabbitmq-server-generic-unix-3.7.9.tar.xz，使用xz -d rabbitmq-server-generic-unix-3.7.9.tar.xz，tar -xvf rabbitmq-server-generic-unix-3.7.9.tar解压。

2.配置环境变量

vim /etc/profile

3.使用守护线程的模式启动rabbitmq

rabbitmq-server -detached

4.添加用户 rabbitmqctl add_user root root123 rabbitmqctl set_permissions -p / root "." "." ".*" rabbitmqctl set_user_tags root administrator

5.虽然已经启动了rabbitmq，但是后台管理还没有打开，需要使用下面的命令打开后台管理。

rabbitmqctl start_app

rabbitmq-plugins enable rabbitmq_management

此时还是有问题，需要关闭防火墙，systemctl stop firewalld.service

此时访问http://192.168.123.124:15672 登录后可以看到如下的界面:

到此为止我们就已经对rabbitmq有了基本认识,也能做些基本安装配置。

简单的例子:

1.导入依赖:

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

消息生产者：

  public class DemoProducer {private static final String EXCHANGE_NAME = "exchange_demo";private static final String ROUNTING_KEY = "routingkey_demo";private static final String QUEUE_NAME = "queue_demo";private static final String IP_ADDRESS = "192.168.124.129";private static final int PORT = 5672;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost(IP_ADDRESS);factory.setPort(PORT);factory.setUsername("root");factory.setPassword("root@123");//创建链接Connection connection = factory.newConnection();//创建信道Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);//创建一个持久化，非排他的，非自动删除的的队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);//将交换器与队列通过路由键绑定channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUNTING_KEY);//发送一条持久化的消息：hello world;String message = "hello world";channel.basicPublish(EXCHANGE_NAME, ROUNTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//关闭资源channel.close();connection.close();}}

消息消费者：

  public class ConsumerDemo {private static final String EXCHANGE_NAME = "exchange_demo";private static final String ROUNTING_KEY = "routingkey_demo";private static final String QUEUE_NAME = "queue_demo";private static final String IP_ADDRESS = "192.168.124.129";private static final int PORT = 5672;public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Address[] addresses = new Address[]{new Address(IP_ADDRESS, PORT)};ConnectionFactory factory = new ConnectionFactory();factory.setHost(IP_ADDRESS);factory.setPort(PORT);factory.setUsername("root");factory.setPassword("root@123");//这里的链接方式和生产者的链接链接方式有所不同，需要区别对待Connection connection = factory.newConnection(addresses);//创建信道final Channel channel = connection.createChannel();//设置客户端最多接收为被ack的消息个数channel.basicQos(64);Consumer consumer = new DefaultConsumer(channel) {[@Override](https://my.oschina.net/u/1162528)public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("recv message:" + new String(body));try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume(QUEUE_NAME, consumer);TimeUnit.SECONDS.sleep(5);channel.close();connection.close();}}

执行结果:

转载于:https://my.oschina.net/u/3474937/blog/3008325