RabbitMQ分析全解之快速开始实现一个demo

RabbitMQ的安装、Web管理、命令行控制

基础操作主要包括：安装服务器，查看Web管理页面，使用控制行操作相关内容。

安装服务器

安装Erlang

RabbitMQ主要依赖Erlang，因此必须先安装erlang。

erlang主要有三个方式来安装。具体内容参考： http://www.rabbitmq.com/install-rpm.html

这里选择第二种进行安装。

安装erlang-solution的配置信息

wget http://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm
rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

运行yum，安装erlang.rpm表

sudo yum install erlang
``` 
3. 测试，运行 erl命令

#### 安装RabbitMQ
1. 我们这里选择手动下载rabbitmq安装包，当然也可以选择rpm和yum进行安装。

wget https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.15/rabbitmq-server-mac-standalone-3.6.15.tar.xz

2. 启动服务和检查服务是否正常启动

./sbin/rabbitmq-plugins enable rabbitmq_management

./sbin/rabbitmq-server -detached #这是后台服务运行

./sbin/rabbitmqctl status #监测服务状态

3. 查看Web管理平台界面

浏览器打开http://localhost:15672/
用户名和密码是 guest／guest


4. 新增用户admin并设置状态administrator

./sbin/rabbitmqctl add_user admin 123456

./sbin/rabbitmqctl set_user_tags admin administrator

[rabbitmq官方参考资料：https://www.rabbitmq.com/management.html#configuration](https://www.rabbitmq.com/management.html#configuration)
### 采用广播模式Fanout实现一个发送接收消息的demo
#### 生产消息
* Maven依赖配置
```xml
     <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter</artifactId>
       </dependency>

       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-web</artifactId>
       </dependency>

       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-test</artifactId>
           <scope>test</scope>
       </dependency>
      <!-- rabbitmq 依赖 -->
       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-amqp</artifactId>
       </dependency>

       <dependency>
           <groupId>com.google.code.gson</groupId>
           <artifactId>gson</artifactId>
       </dependency>

生产消息配置

spring.application.name=springboot-rabbitmq-app-provider
server.port=8084

#rabbitmq相关配置
spring.rabbitmq.host = 127.0.0.1
spring.rabbitmq.port = 5672
spring.rabbitmq.username = admin
spring.rabbitmq.password = 123456


# 自定义配置应用于消息生产
#交换器名称
rabbitmq.config.exchange = mc.Ex.demo.test
#自定义队列名称
rabbitmq.producer.queue.name =mc.queue.demo.test

生产消息Rabbitmq配置

/**
 * @author vincent.li
 * @Description Rabbitmq配置
 * @since 2019/8/10
 */
@Component
public class RabbitmqConfig {

    @Value("${rabbitmq.config.exchange}")
    private String fanoutExchangeName;
    @Value("${rabbitmq.producer.queue.name}")
    private String producerQueueName;

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(fanoutExchangeName, true, false);
    }

    @Bean(name = "basicQueue")
    public Queue basicQueue() {
        return new Queue(producerQueueName, true);
    }

    @Bean
    public Binding basicBinding() {
        return BindingBuilder.bind(basicQueue()).to(fanoutExchange());

    }
}

生产消息方法

/**
 * @author vincent.li
 * @Description Rabbitmq消息生产控制器
 * @since 2019/8/9
 */
@Slf4j
@RestController
@RequestMapping(value = "rabbitmq")
public class RabbitmqProduceController {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${rabbitmq.config.exchange}")
    private String fanoutExchangeName;

    /**
     * 发送消息
     * @param msg
     * @return
     */
    @GetMapping("/message/send")
    public BaseResponse sendMessage(@RequestParam String msg){
        if (StringUtils.isEmpty(msg)) {
            return BaseResponse.fail("请输入有效参数");
        }
        try {
            log.info("待发送的消息： {} ", msg);

            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setExchange(fanoutExchangeName);

            Message message = Message.builder()
                    .id(System.currentTimeMillis())
                    .msg(msg)
                    .sendTime(new Date())
                    .build();
            rabbitTemplate.convertAndSend(message);
        }catch (Exception e){
            log.error("发送消息发生异常： ", e.fillInStackTrace());
            return BaseResponse.fail("发送消息发生异常");
        }
        return BaseResponse.success();
    }



}

接收消息

Maven依赖配置

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
  </dependency>

  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
  </dependency>

  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
  </dependency>
  <!-- rabbitmq 依赖 -->
  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

  <dependency>
      <groupId>com.google.code.gson</groupId>
      <artifactId>gson</artifactId>
  </dependency>

接收消息配置

spring.application.name=springboot-rabbitmq-app-consumer
server.port=8083

spring.rabbitmq.host = 127.0.0.1
spring.rabbitmq.port = 5672
spring.rabbitmq.username = admin
spring.rabbitmq.password = 123456

# 自定义配置应用于消息消费
#交换器名称
rabbitmq.config.exchange = mc.Ex.demo.test

#自定义队列名称
rabbitmq.consumer.queue.name =mc.queue.demo.test.consume

rabbitmq.consumer.2.queue.name = mc.queue.demo.test.consume2

接收消息consumer-1

/**
 * @author vincent.li
 * @Description rabbitmq广播模式消费
 * @since 2019/8/9
 */
@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "${rabbitmq.consumer.queue.name}", durable = "true", autoDelete = "false"),
        exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.FANOUT))
)
public class RabbitmqFanoutConsumer {

    /**
     * 设置监听方法
     *
     * @param message
     * @RabbitHandler 声明监听方法是下面的 isDefault属性是默认false接受的完整对象，true接受body体
     */
    @RabbitHandler(isDefault = true)
    public void process(byte[] message) {
        System.out.println("Fanout-1 Receiver : " + new String(message));
    }

}

接收消息consumer-2

/**
 * @author vincent.li
 * @Description rabbitmq广播模式消费
 * @since 2019/8/9
 */
@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "${rabbitmq.consumer.2.queue.name}", durable = "true", autoDelete = "false"),
        exchange = @Exchange(value = "${rabbitmq.config.exchange}", type = ExchangeTypes.FANOUT))
)
public class RabbitmqFanoutConsumer2 {

    /**
     * 设置监听方法
     *
     * @param message
     * @RabbitHandler 声明监听方法是下面的 isDefault属性是默认false接受的完整对象，true接受body体
     */
    @RabbitHandler(isDefault = true)
    public void process(byte[] message) {
        System.out.println("Fanout-2 Receiver : " + new String(message));
    }

}

运行结果

发送消息：

发送消息

接受消息：

接受图片

具体完整代码可以移步github:springcloud-app-rabbitmq