RabbitMQ安装及简单应用

代码小搬运 2019-12-02

121

安装：

下载地址：http://www.rabbitmq.com/install-standalone-mac.html
解压：tar zxvf rabbitmq-server-mac-standalone-3.7.14.tar.xz

基本操作命令：切换到RabbitMQ安装目录的sbin目录下，使用操作命令

启动RabbitMQ：

./rabbitmq-server restart 此命令启动不能关闭终端，关闭终端则RabbitMQ服务断开
./rabbitmq-server -detached 此命令启动关闭终端则不会影响到RabbitMQ服务

停止 RabbitMQ：./rabbitmqctl stop

可通过访问 http://localhost:15672 进行测试，默认登录账户为：guest guest

以上若不能正常访问RabbitMQ界面，则需要首先启动其管理模块

开启RabbitMQ节点：./rabbitmqctl start_app
开启RabbitMQ管理模块的插件，并配置到RabbitMQ节点上：./rabbitmq-plugins enable rabbitmq_management
关闭RabbitMQ节点：./rabbitmqctl stop

简单使用：

maven依赖

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

配置文件

spring:
  application:
    name: spring-boot-rabbitmq
  rabbitmq:
    # 连接地址
    host: 127.0.0.1
    # 端口号，此为默认端口
    port: 5672
    # 连接用户名，默认
    username: guest
    # 连接密码，默认
    password: guest

消息队列

@Configuration
public class RabbitConfig {


    @Bean
    public Queue queue() {
        //此处定义的队列名称，当此配置文件加载后此队列会加载至RabbitMQ客户端
        return new Queue("hellos");
    }
}

发送者

@Slf4j
@Component
public class HelloSender {


    @Autowired
    AmqpTemplate amqpTemplate;


    public void send(String messgae) {
        log.info("send message: {}", messgae);
        //定义的队列名称需在RabbitMQ服务器中存在，需与接收者一致
        amqpTemplate.convertAndSend("hello", messgae);
    }
}

接收者(下面配置了两个接收者)

@Slf4j
@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {


    @RabbitHandler
    public void process(String messge) {
        log.info("===============HelloReceiver receiver message: {}", messge);
    }
}


@Slf4j
@Component
//接收消息的队列名称
@RabbitListener(queues = "hello")
public class HelloReceiver2 {


    @RabbitHandler
    public void process(String messge) {
        log.info("===============HelloReceiver2 receiver message: {}", messge);
    }
}

测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {


    @Autowired
    private HelloSender helloSender;


    /**
     * 单个测试
     *
     * @throws Exception
     */
    @Test
    public void hello() throws Exception {
        helloSender.send("测试");
    }


    /**
     * 多消费者
     *
     * @throws Exception
     */
    @Test
    public void oneToMany() throws Exception {
        for (int i = 1; i <= 10; i++) {
            helloSender.send(String.valueOf(i));
        }
    }
}

输出：

使用交换机（Exchange）：交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct(direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去), Topic(按规则转发消息（最灵活）), Headers(设置 header attribute 参数类型的交换机)，Fanout(转发消息到所有绑定队列)

Direct Exchange：需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog，任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。
Topic Exchange：

规则配置

@Configuration
public class TopicRabbitConfig {


   final static String message = "topic.message";
   final static String messages = "topic.messages";


   @Bean
   public Queue queueMessage() {
       return new Queue(TopicRabbitConfig.message);
   }


   @Bean
   public Queue queueMessages() {
       return new Queue(TopicRabbitConfig.messages);
   }


   @Bean
   TopicExchange exchange() {
       return new TopicExchange("exchange");
   }


   @Bean
   Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
       //将routing_key为topic.message的通过交换机绑定到queueMessage队列
       return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
   }


   @Bean
   Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
       //将routing_key已topic.开头的通过交换机绑定到queueMessages队列
       return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
   }
}

发送者：通过交换机绑定不同的routing_key来实现到达不同的队列

@Slf4j
@Component
public class TopicRabbitSender {


    @Autowired
    RabbitTemplate rabbitTemplate;


    public void send1(User user){
        //exchange：交换机的名称；routingKey：routing_key；object：发送的数据
        rabbitTemplate.convertAndSend("exchange", "topic.message",user);
    }


    public void send2(User user){
        rabbitTemplate.convertAndSend("exchange", "topic.messages",user);
    }
}

接收者

@Slf4j
@Component
@RabbitListener(queues = "topic.message")
public class TopicRabbitReceiver1 {


    @RabbitHandler
    public void process(User user) {
        log.info("=========queue topic.message receiver :{}", user.toString());
    }
}


@Slf4j
@Component
@RabbitListener(queues = "topic.messages")
public class TopicRabbitReceiver2 {


    @RabbitHandler
    public void process(User user) {
        log.info("=========queue topic.messages receiver :{}", user.toString());
    }
}

测试：

  @Test
    public void topic() {
        for (int i = 1; i <= 10; i++) {
            topicRabbitSender.send1(new User(Long.valueOf(i), "a"));
        }


        for (int i = 1; i <= 10; i++) {
            topicRabbitSender.send2(new User(Long.valueOf(i),"b" ));
        }
    }

打印测试结果：

接收方式与普通的接收方式是一致的，区别在于匹配队列

Fanout Exchange：广播模式或者订阅模式，给 Fanout 交换机发送消息，绑定了这个交换机的所有队列都收到这个消息：

规则配置

@Configuration
public class FanoutRabbitConfig {


    @Bean
    public Queue fanoutA(){
        return new Queue("fanoutA");
    }


    @Bean
    public Queue fanoutB(){
        return new Queue("fanoutB");
    }


    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanoutExchange");
    }


    @Bean
    Binding bindingFanoutExchangeA(FanoutExchange fanoutExchange,Queue fanoutA){
        //将队列fanoutA与fanoutExchange交换机进行绑定
        return BindingBuilder.bind(fanoutA).to(fanoutExchange);
    }


    @Bean
    Binding bindingFanoutExchangeB(FanoutExchange fanoutExchange,Queue fanoutB){
        //将队列fanoutB与fanoutExchange交换机进行绑定
        return BindingBuilder.bind(fanoutB).to(fanoutExchange);
    }
}

发送者：

@Slf4j
@Component
public class FanoutRabbitSender {


    @Autowired
    RabbitTemplate rabbitTemplate;


    public void send(String message) {
        //exchange：交换机的名称；routingKey：Fanout交换机无法使用routing_key，所以这里不设置routing_key；object：发送的数据
        rabbitTemplate.convertAndSend("fanoutExchange", "", message);
    }
}

接收者

@Slf4j
@Component
@RabbitListener(queues = "fanoutA")
public class FanoutRabbitReceiver1 {


    @RabbitHandler
    public void process(String message) {
        log.info("=========queue fanoutA receiver :{}", message);
    }
}


@Slf4j
@Component
@RabbitListener(queues = "fanoutB")
public class FanoutRabbitReceiver2 {


    @RabbitHandler
    public void process(String message) {
        log.info("=========queue fanoutB receiver :{}", message);
    }
}

测试：因为绑定的交换机一致，不像Topic交换机可通过routing_key区分队列，所以只要绑定此交换机的队列都会接收消息

  @Test
    public void fanout(){
        for (int i = 1; i <= 5; i++) {
            fanoutRabbitSender.send("测试Fanout交换机"+i);
        }
    }

数据库

文章转载自代码小搬运，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

RabbitMQ安装及简单应用

评论