P：代表生产者，向队列中发送消息。

队列:保存生产者提供的数据。(队列的特点，先进先出)

C1,C2:代表消费者,负责将对列中的消息全部读取，并且完成特定的任务。

工作原理:当生产者向对列中发送消息时，C1和C2一起争抢消息的执行权，谁抢到，谁执行。内部有负载的策略。

[root@bogon rabbitmq-server-3.6.1]# cd /etc/rabbitmq

[root@bogon rabbitmq]# service rabbitmq-server start

Starting rabbitmq-server: SUCCESS

rabbitmq-server.er.

pom.xml

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

com.rabbit

schoolmanage

0.0.1-SNAPSHOT

war

com.rabbitmq

amqp-client

3.5.1

org.springframework.amqp

spring-rabbit

1.4.0.RELEASE

junit

junit

4.11

package schoolmanage;

import java.io.IOException;

import org.junit.After;

import org.junit.Before;

import org.junit.Test;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.QueueingConsumer;

public class TestStudentMsgWork {

private Connection connection = null;

// 工作模式

private String queueName = "work";

@Before

public void init() throws IOException {

// 创建连接工厂

ConnectionFactory factory = new ConnectionFactory();

// 为工厂对象添加数据

// 远程主机

factory.setHost("192.168.6.130");

// 端口号

factory.setPort(5672);

// 虚拟主机

factory.setVirtualHost("/school");

// 用户名

factory.setUsername("student");

// 密码

factory.setPassword("student");

// 创建连接

connection = factory.newConnection();

}

// 消息生产者

@Test

public void provider() throws IOException {

// 创建通道

Channel channel = connection.createChannel();

String msg = "我要吃饭!!!——工作模式";

// 创建队列

/**

* String queue, 队列的名称 boolean durable,持久化 false表示不持久化,

* true表示rabbitmq重启后会恢复队列的内容 boolean exclusive,服务器所独有,如果设置为true则消费者不能使用

* boolean autoDelete,是否自动删除 当队列中没有消息时，该队列是否自动删除 Map

* arguments 额外的参数

*/

channel.queueDeclare(queueName, false, false, false, null);

/**

* String exchange, 交换机的名称,如果需要交换机则添加名称 如果没有交换机 则为""串 String routingKey,

* 路由key 寻址的关键字，如果需要使用路由key定义特定的关键字(orderKey.xxx)

* 如果不需要路由key,在简单模式中的添加队列的名称 BasicProperties props, 其他的配置,一般为null byte[]

* body,表示需要发送的消息-字节码文件

*/

channel.basicPublish("", queueName, null, msg.getBytes());

// 将流关闭

channel.close();

System.out.println("消息发送成功!!!");

}

// 消费者1

@Test

public void consumer1() throws Exception {

// 创建通道

Channel channel = connection.createChannel();

channel.queueDeclare(queueName, false, false, false, null);

// 需要对消费个数进行定义

// 消费者允许有三次没有响应给rabbitMQ,如果长时间没有响应则不再允许再次消费消息

channel.basicQos(1);// 当前执行的总数，一个消费者一次只能拿到一个队列

// 定义消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 将消费者与队列进行绑定

/**

* autoAck为false表示手动返回

*/

channel.basicConsume(queueName, false, consumer);

System.out.println("消费者1,启动。。。");

// 获取消息

while (true) {

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String msg = "消费者1收到 :" + new String(delivery.getBody());

System.out.println(msg);

// 告知rabbitMq当前消费的是哪一个消息

/**

* multiple false表示不扩展

*/

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

}

}

// 消费者2

@Test

public void consumer2() throws Exception {

// 创建通道

Channel channel = connection.createChannel();

channel.queueDeclare(queueName, false, false, false, null);

// 需要对消费个数进行定义

// 消费者允许有三次没有响应给rabbitMQ,如果长时间没有响应则不再允许再次消费消息

channel.basicQos(1);// 当前执行的总数，一个消费者一次只能拿到一个队列

// 定义消费者

QueueingConsumer consumer = new QueueingConsumer(channel);

// 将消费者与队列进行绑定

/**

* autoAck为false表示手动返回

*/

channel.basicConsume(queueName, false, consumer);

System.out.println("消费者2,启动。。。");

// 获取消息

while (true) {

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String msg = "消费者2收到 :" + new String(delivery.getBody());

System.out.println(msg);

// 告知rabbitMq当前消费的是哪一个消息

/**

* multiple false表示不扩展

*/

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

}

}

@After

public void close() throws IOException {

connection.close();

}

}

执行结果

启动consumer1

消费者1,启动。。。

启动consumer2

消费者2,启动。。。

启动provider

消息发送成功!!!

查看consumer1

消费者1收到 :我要吃饭!!!——工作模式

启动provider

消息发送成功!!!

查看启动consumer2

消费者2收到 :我要吃饭!!!——工作模式