1，安装
测试服务器centos 6.7 ，分别下载erlang-20.3.8.21-1.el6.x86_64.rpm，rabbitmq-server-3.7.14-1.el6.noarch.rpm，socat-1.7.3.2-1.1.el6.x86_64.rpm对应版本，使用rpm命令进行安装。
2，安装完成后，执行如下命令：
chkconfig rabbitmq-server on
service rabbit-server start
启动rabbitmq。如果出现启动异常，crash的情况，确认是否存在端口占用的情况。
3，默认会创建一个名为guest密码为guest的用户，只用在本地通过localhost进行连接，如果需要在其他服务器上连接需要进行access control设置。创建测试用户
rabbitmqctl add_user rabbitmq 12345678
rabbitmqctl set_permissions rabbitmq ".*" ".*" ".*"
rabbitmqctl set_user_tags rabbitmq administrator
rabbitmqctl delete_user guest
AMQP是一个用来处理发布者和消费者的消息协议。publishe发布消息，consumer选取消息并处理，broker的作用就是保证从发布者发布的消息能到达消费者。broker的关键组件是exchanges和queues（图片来源https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-by-rabbitmq/）。
如图所示，发布者发布消息到指定的exchanges，然后消费者从queue中拉取消息或queue向消费者推送消息。发布者和消费者通过exchange发现彼此。exchange的创建根据实际情况，要么由发布者创建要么由消费者创建。
消息从exchange到queue的路由过程，queue要依附于一个指定的exchange。一般情况下，消费者创建一个queue同时依附于一个exchange。由exchange接收的消息会匹配到对应的queue，这个过程叫做binding。
AMQP的消息由三部分组成，headers,properties，byte[] data。其中headers和properties是key-value形式的数据。消息体就是字节数组，如果发送文本信息需要指定编码。
headers中有一个routing-key的属性，broker用这个属性将消息匹配到指定的queue，每个queue也会有一个binding key，如果routing-key和binding key匹配，则queue就会接受这个消息。
exchange有四种类型：direct（binding key和routing-key全匹配），topic（类似于direct，但是可以使用wildcards），fanout（忽略binding key 和routing-key），headers。
基于以上对于rabbitmq的理解来使用spring实现简单的发送消息和消费消息。
1，添加起步配置
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2，
@SpringBootApplication
@EnableScheduling
@EnableRabbit
public class DemoForRabbitmq {
....
}
3，配置ConnectionFactory
@Bean
public ConnectionFactory connectionFactory(){
CachingConnectionFactory factory = new CachingConnectionFactory("ip",5672);
factory.setUsername("rabbitmq");
factory.setPassword("12345678");
factory.setChannelCacheSize(50);
return factory;
}
4，配置amqpAdmin
@Bean
public AmqpAdmin amqpAdmin(){
RabbitAdmin amqpAdmin = new RabbitAdmin(connectionFactory());
return amqpAdmin;
}
5，配置Exchange
@Bean
public DirectExchange dex1(){
return new DirectExchange("dex1");
}
@Bean
public TopicExchange tex1(){
return new TopicExchange("tex1");
}
6，配置queue
@Bean
public Queue firstQueue(){
return new Queue("firstQueue");
}
@Bean
public Queue secondQueue(){
return new Queue("secondQueue");
}
7，配置binding
@Bean
public Binding b1(){
return BindingBuilder.bind(firstQueue()).to(dex1()).with("key1");
}
@Bean
public Binding b2(){
return BindingBuilder.bind(secondQueue()).to(tex1()).with("key2");
}
8，使用declarables维护exchange，queue和bindings，该配置会有amqpAdmin对象进行维护
@Bean
public Declarables bs(){
return new Declarables(new DirectExchange("dex2"),
new Queue("thirdQueue"),
new Binding("thirdQueue",DestinationType.QUEUE,"dex2","key3",null));
}
9，配置RabbitTamplate
@Bean
public RabbitTemplate amqpTemplate(){
RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory());
RetryTemplate retryTemplate = new RetryTemplate();
ExponentialRandomBackOffPolicy backOffPolicy = new ExponentialRandomBackOffPolicy();
backOffPolicy.setInitialInterval(500);
backOffPolicy.setMultiplier(10.0);
backOffPolicy.setMaxInterval(10000);
retryTemplate.setBackOffPolicy(backOffPolicy);
amqpTemplate.setMessageConverter(messageConverter());
amqpTemplate.setRetryTemplate(retryTemplate);
return amqpTemplate;
}
10，配置MessageConverter
@Bean
public MessageConverter messageConverter(){
return new SimpleMessageConverter();
}
11，监听Queue
@Bean
public SimpleMessageListenerContainer container(){
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
//reply 设置
// container.setMessageListener(amqpTemplate());
container.setQueueNames("thirdQueue");
container.setMessageListener(message -> System.out.println("received message from thirdQueue" +message));
return container;
}
12，创建消息发布者
@Component
public class Producer4Amqp{
@Scheduled(fixedDelay = 3000L)
public void send() {
System.err.println("sending message"+atomicInteger.get());
if(atomicInteger.get()%3 ==0){
this.amqpTemplate.send("dex1","key1",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
}else if(atomicInteger.get()%3 ==1){
this.amqpTemplate.send("tex1","key2",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
}else if(atomicInteger.get()%3 ==2){
this.amqpTemplate.send("dex2","key3",MessageBuilder.withBody((atomicInteger.incrementAndGet()+" message").getBytes()).
setContentEncoding(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).setMessageId(atomicInteger.get()+"").build());
}
}
}
13，创建消息消费者
@Component
public class Consumer4Amqp {
@RabbitListener(queues = {"firstQueue"})
public void received(byte[] message){
System.out.println("consumer 1 received message from firstQueue" + new String(message));
}
@RabbitListener(queues = "secondQueue")
public void recevied2(Message message){
System.out.println("consumer 2 received message from secondQueue" + message);
}
}
启动程序，从console里可以看到如下日志信息：
consumer 1 received message from firstQueue85 message
sending message85
consumer 2 received message from secondQueue(Body:'[B@551d5210(byte[10])' MessageProperties [headers={}, messageId=86, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=tex1, receivedRoutingKey=key2, deliveryTag=39, consumerTag=amq.ctag-cHDXBlrMu0bOgZ4hJSCKbA, consumerQueue=secondQueue])
sending message86
received message from thirdQueue(Body:'[B@585fb560(byte[10])' MessageProperties [headers={}, messageId=87, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=dex2, receivedRoutingKey=key3, deliveryTag=39, consumerTag=amq.ctag-1S_ttl_zj7VdbYHjPScViA, consumerQueue=thirdQueue])
sending message87
consumer 1 received message from firstQueue88 message
sending message88
consumer 2 received message from secondQueue(Body:'[B@52121840(byte[10])' MessageProperties [headers={}, messageId=89, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=tex1, receivedRoutingKey=key2, deliveryTag=40, consumerTag=amq.ctag-cHDXBlrMu0bOgZ4hJSCKbA, consumerQueue=secondQueue])
sending message89
received message from thirdQueue(Body:'[B@451add0(byte[10])' MessageProperties [headers={}, messageId=90, contentType=application/octet-stream, contentEncoding=text/plain, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=dex2, receivedRoutingKey=key3, deliveryTag=40, consumerTag=amq.ctag-1S_ttl_zj7VdbYHjPScViA, consumerQueue=thirdQueue])
sending message90
consumer 1 received message from firstQueue91 message
可以使用rabbitmqctl list_exchanges/list_queus/list_consumers等查看队列状态。
