rabbitmq

2019-04-08 本文已影响0人原来_1361

关于rabbitmq的用法参考：https://www.cnblogs.com/linkenpark/p/5393666.html

一、配置

1. Exchange

exchange是一个路由器，消息是经过路由器投递到对应的队列中。

/**
 * 消息交换机配置  可以配置多个
 *
 * @Author administrator
 * @date 2019/4/5
 */
@Configuration
public class ExchangeConfig {

    /**
     * direct exchange: routing key完全匹配才转发
     *
     * @return
     */
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(RabbitMqConfig.EXCHANGE, true, false);
    }

    /**
     * fanout exchange: 不理会routing key,消息直接广播到所有绑定的queue
     *
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(RabbitMqConfig.EXCHANGE, true, false);
    }

    /**
     * topic exchange : 对routing key模式匹配
     *
     * @return
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(RabbitMqConfig.EXCHANGE, true, false);
    }
}

2. Queue

队列，存放消息的载体。

@Configuration
public class QueueConfig {
    @Bean
    public Queue queue(){
        return new Queue("whc");
    }
}

3. RouteKey

映射规则，根据routekey将队列与路由器绑定。

package com.whc.config;

import com.whc.callback.MsgSendConfirmCallBack;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 *
 */
@Configuration
public class RabbitMqConfig {
    /**
     * 消息交换机的名字
     */
    public static final String EXCHANGE = "exchange_test";
    /**
     * 队列key1
     */
    public static final String ROUTE_KEY_1 = "query_one_key1";
    /**
     * 队列key2
     */
    public static final String ROUTE_KEY_2 = "query_one_key2";

    @Autowired
    private QueueConfig queueConfig;
    @Autowired
    private ExchangeConfig exchangeConfig;

    /**
     * 连接工厂
     */
    @Autowired
    private ConnectionFactory factory;

    /**
     * 将消息队列和交换机通过路由key绑定
     *
     * @return
     */
    @Bean
    public Binding bindingOne() {
        return BindingBuilder.bind(queueConfig.queue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTE_KEY_1);
    }

    /**
     * queue监听器，观察者模式
     * 当有消息到达时会通知监听在对应队列上的对象
     *
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(factory);
        container.addQueues(queueConfig.queue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(5);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return container;
    }

    /**
     * 消息确认机制
     * Confirms给客户端一种轻量级的方式，能够跟踪哪些消息被broker处理，
     * 哪些可能因为broker宕掉或者网络失败的情况而重新发布。
     * 确认并且保证消息被送达，提供了两种方式：发布确认和事务。(两者不可同时使用)
     * 在channel为事务时，不可引入确认模式；同样channel为确认模式下，不可使用事务
     *
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(factory);
        rabbitTemplate.setConfirmCallback(msgSendConfirmCallBack());
        return rabbitTemplate;
    }

    @Bean
    public MsgSendConfirmCallBack msgSendConfirmCallBack() {
        return new MsgSendConfirmCallBack();
    }
}

4. pom依赖和配置文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.whc</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq</name>
    <description>rabitmq for springboot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--rabbit-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.properties

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

二、生产者和消费者

FirstConsumer

package com.whc.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 这里有两个消费者FirstConsumer和HelloReceiver,都订阅了whc这个消息队列，消息消费的原则是：只能由一个消费者消费
 *
 * @author Administrator
 * @date 2019/4/8
 */
@Component
public class FirstConsumer {
    @RabbitListener(queues = {"whc"}, containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(String message) throws Exception {
        System.out.println("FirstConsumer {} handle message..." + message);
    }
}

package com.whc.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @ClassName HelloReceiver
 * @Description TODO
 * @Author Administrator
 * @Date 2018/12/8 19:48
 * @Version 1.0
 */
@Component
@RabbitListener(queues = "whc")
public class HelloReceiver {

    @RabbitHandler
    public void process(String hello) {
        System.err.println("Receiver  : " + hello);
    }
}

package com.whc.sender;

import com.whc.config.RabbitMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author Administrator
 * @date 2019/4/8
 */
@Slf4j
@Component
public class FirstSender {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param uuid
     * @param msg
     */
    public void send(String uuid, Object msg){
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE, RabbitMqConfig.ROUTE_KEY_1, msg, correlationId);
    }

}

package com.whc.sender;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @ClassName HelloSender
 * @Description TODO
 * @Author Administrator
 * @Date 2018/12/8 19:47
 * @Version 1.0
 */
@RestController
public class HelloSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    /**
     * 发送单条信息
     */
    @GetMapping("/send")
    public void send() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String context = "hello " + sdf.format(new Date());
        System.out.println("Sender : " + context);
        //通过rabbitTemplate.convertAndSend发送context到whc队列中
        this.rabbitTemplate.convertAndSend("whc", context);
    }

    /**
     * 循环发送多条信息
     */
    @GetMapping("/multiSend")
    public void multiSend() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        for (int i = 0; i < 1000; i++) {
            String context = "你好，这是我的第 " + i + "条消息，当前时间为：" + sdf.format(new Date());
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("whc", context);
        }
    }
}

三、测试用例

package com.whc;

import com.whc.sender.FirstSender;
import com.whc.sender.HelloSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.UUID;

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

    @Autowired
    private HelloSender helloSender;

    @Test
    public void contextLoads() {
    }

    @Test
    public void hello() throws Exception {
        helloSender.send();
    }

    @Autowired
    private FirstSender firstSender;

    @Test
    public void send() throws InterruptedException {
        String uuid = UUID.randomUUID().toString();
        for (int i = 0; i < 100; ) {
            firstSender.send(uuid, "hello world, this id send by the " +(++i)+ " one...");
            System.out.println("I am sending you " + i + " message.");
            Thread.sleep(100);
        }

    }
}

rabbitmq

一、配置

1. Exchange

2. Queue

3. RouteKey

4. pom依赖和配置文件

二、生产者和消费者

三、测试用例

猜你喜欢

热点阅读

rabbitmq

一、 配置

1. Exchange

2. Queue

3. RouteKey

4. pom依赖和配置文件

二、 生产者和消费者

三、 测试用例

猜你喜欢

热点阅读

一、配置

二、生产者和消费者

三、测试用例