zoukankan      html  css  js  c++  java
  • 88--spring cloud (Springboot 整合RabbitMQ)

    RabbitMQ - Spring boot 整合

    新建项目

    导入依赖RabbtiMQ

    pom.xml文件

    <?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <groupId>com.jt</groupId>
        <artifactId>springBootDemo-rabbitmq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>springBootDemo-rabbitmq</name>
        <description>Demo project for Spring Boot</description>
    
        <properties>
            <java.version>1.8</java.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <spring-boot.version>2.3.0.RELEASE</spring-boot.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
                <exclusions>
                    <exclusion>
                        <groupId>org.junit.vintage</groupId>
                        <artifactId>junit-vintage-engine</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-dependencies</artifactId>
                    <version>${spring-boot.version}</version>
                    <type>pom</type>
                    <scope>import</scope>
                </dependency>
            </dependencies>
        </dependencyManagement>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.1</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                    </configuration>
                </plugin>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <version>2.3.0.RELEASE</version>
                    <configuration>
                        <mainClass>com.jt.SpringBootDemoRabbitmqApplication</mainClass>
                    </configuration>
                    <executions>
                        <execution>
                            <id>repackage</id>
                            <goals>
                                <goal>repackage</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    
    </project>
    
    

    application.yml配置

    spring:
      rabbitmq:
        host: 120.53.24.172
        username: admin
        password: admin
    

    简单模式

    主程序

    Spring提供的Queue类,是队列的封装对象,它封装了队列的参数信息.

    RabbitMQ的自动配置类,会发现这些Queue实例,并在RabbitMQ服务器中定义这些队列.

    package com.jt;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    
    @SpringBootApplication
    public class SpringBootDemoRabbitmqApplication {
    
        public static void main(String[] args) {
            SpringApplication.run(SpringBootDemoRabbitmqApplication.class, args);
        }
        @Bean
    	public Queue task_queue() {
    		/*
    		 * 可用以下形式: 
    		 * new Queue("helloworld") - 持久,非排他,非自动删除
    		 * new Queue("helloworld",false,false,false,null)
    		 */
    		return new Queue("helloworld",false);
    	}
    
    }
    
    

    生产者

    AmqpTemplate是rabbitmq客户端API的一个封装工具,提供了简便的方法来执行消息操作.

    AmqpTemplate由自动配置类自动创建

    package com.jt.m1;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class SimpleSender {
    	@Autowired
    	AmqpTemplate t;
    	
    	public void send() {
    		// 这里向 helloworld 队列发送消息
    		t.convertAndSend("helloworld", "Hello world!! "+System.currentTimeMillis());
    		System.out.println("消息已发送");
    	}
    
    }
    
    

    消费者

    通过@RabbitListener从指定的队列接收消息

    使用@RebbitHandler注解的方法来处理消息

    package com.jt.m1;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    @Component
    @RabbitListener(queues = "helloworld")
    public class SimpleReceiver {
    	@RabbitHandler
    	public void receive(String msg) {
    		System.out.println("收到: "+msg);
    	}
    	
    }
    

    这里还可以使用另一种形式:

    @Component
    public class SimpleReceiver {
    	@RabbitListener(queues = "helloworld")
    	public void receive(String msg) {
    		System.out.println("收到: "+msg);
    	}
    }
    
    

    另外,@RabbitListener 注解中也可以直接定义队列:

    	@RabbitListener(queuesToDeclare = @Queue(name = "helloworld",durable = "false"))
    
    

    测试类

    在存放测试代码的目录中,创建测试类

    package com.jt.m1;
    
    import java.util.Scanner;
    
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    @SpringBootTest
    public class SimpleTests {
    	@Autowired
    	SimpleSender simpleSender;
    
    	@Test
    	void test1() throws Exception {
    		while(true) {
    			
    			simpleSender.send();
    			System.out.println("[按回车结束]");
    			new Scanner(System.in).nextLine();
    		}
    	}
    
    }
    
    

    工作模式

    主程序

    在主程序中创建名为task_queue持久队列

    package com.jt;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    
    @SpringBootApplication
    public class SpringBootDemoRabbitmqApplication {
    
        public static void main(String[] args) {
            SpringApplication.run(SpringBootDemoRabbitmqApplication.class, args);
        }
        @Bean
    	public Queue task_queue() {
    		// 这个构造方法创建的队列参数为: 持久,非排他,非自动删除
    		return new Queue("task_queue");
    	}
    //    @Bean
    //	public Queue task_queue() {
    //		/*
    //		 * 可用以下形式: 
    //		 * new Queue("helloworld") - 持久,非排他,非自动删除
    //		 * new Queue("helloworld",false,false,false,null)
    //		 */
    //		return new Queue("helloworld",false);
    //	}
    
    }
    
    

    生产者

    package com.jt.m2;
    
    import java.util.Scanner;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class WorkSender {
    	@Autowired
    	AmqpTemplate t;
    	
    	public void send() {
    		while (true) {
    			System.out.print("输入:");
    			String s = new Scanner(System.in).nextLine();
    			
    			//spring 默认将消息的 DeliveryMode 设置为 PERSISTENT 持久化,
    			t.convertAndSend("task_queue", s);
    		}
    	}
    
    }
    
    

    spring boot封装的 rabbitmq api 中, 发送的消息默认是持久化消息.
    如果希望发送非持久化消息, 需要在发送消息时做以下设置:

    • 使用 MessagePostProcessor 前置处理器参数
    • 从消息中获取消息的属性对象
    • 在属性中把 DeliveryMode 设置为非持久化
    	//如果需要设置消息为非持久化,可以取得消息的属性对象,修改它的deliveryMode属性
    	t.convertAndSend("task_queue", (Object) s, new MessagePostProcessor() {
    		@Override
    		public Message postProcessMessage(Message message) throws AmqpException {
    			MessageProperties props = message.getMessageProperties();
    			props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
    			return message;
    		}
    	});
    

    消费者

    ack模式

    在 spring boot 中提供了三种确认模式:

    • NONE - 使用rabbitmq的自动确认
    • AUTO - 使用rabbitmq的手动确认, springboot会自动发送确认回执 (默认)
    • MANUAL - 使用rabbitmq的手动确认, 且必须手动执行确认操作

    默认的 AUTO 模式中, 处理消息的方法抛出异常, 则表示消息没有被正确处理, 该消息会被重新发送.

    设置 ack 模式

    spring:
      rabbitmq:
        listener:
          simple:
            # acknowledgeMode: NONE # rabbitmq的自动确认
            acknowledgeMode: AUTO # rabbitmq的手动确认, springboot会自动发送确认回执 (默认)
            # acknowledgeMode: MANUAL # rabbitmq的手动确认, springboot不发送回执, 必须自己编码发送回执
    

    手动执行确认操作

    如果设置为 MANUAL 模式,必须手动执行确认操作

    	@RabbitListener(queues="task_queue")
    	public void receive1(String s, Channel c, @Header(name=AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    		System.out.println("receiver1 - 收到: "+s);
    		for (int i = 0; i < s.length(); i++) {
    			if (s.charAt(i) == '.') {
    				Thread.sleep(1000);
    			}
    		}
    		// 手动发送确认回执
    		c.basicAck(tag, false);
    	}
    

    抓取数量

    工作模式中, 为了合理地分发数据, 需要将 qos 设置成 1, 每次只接收一条消息, 处理完成后才接收下一条消息.

    spring boot 中是通过 prefetch 属性进行设置, 改属性的默认值是 250.

    spring:
      rabbitmq:
        listener:
          simple:
            prefetch: 1 # qos=1, 默认250
    

    消费者代码

    package com.jt.m2;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.stereotype.Component;
    
    import com.rabbitmq.client.Channel;
    
    
    @Component
    public class WorkReceiver1 {
    	@RabbitListener(queues="task_queue")
    	public void receive1(String s, Channel c, @Header(name=AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    		System.out.println("receiver1 - 收到: "+s);
    		for (int i = 0; i < s.length(); i++) {
    			if (s.charAt(i) == '.') {
    				Thread.sleep(1000);
    			}
    		}
    		c.basicAck(tag,false);
    	}
    	
    	@RabbitListener(queues="task_queue")
    	public void receive2(String s, Channel c, @Header(name=AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
    		System.out.println("receiver2 - 收到: "+s);
    		for (int i = 0; i < s.length(); i++) {
    			if (s.charAt(i) == '.') {
    				Thread.sleep(1000);
    			}
    		}
    		c.basicAck(tag,false);
    	}
    	
    }
    

    发布和订阅模式

    主程序

    创建 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.

    spring boot 的自动配置类会自动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.

    package cn.tedu.rabbitmq.m3;
    
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    
    @SpringBootApplication
    public class Main {
        public static void main(String[] args) {
            SpringApplication.run(Main.class,args);
        }
    
    
        //定义一个交换机
        @Bean
        public FanoutExchange LogsExchange(){
            return  new FanoutExchange("logs",false,false);//非持久 不自动删除
    
        }
    }
    
    

    生产者

    生产者向指定的交换机 logs 发送数据.

    不需要指定队列名或路由键, 即使指定也无效, 因为 fanout 交换机会向所有绑定的队列发送数据, 而不是有选择的发送.

    package cn.tedu.rabbitmq.m3;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Scanner;
    //发布订阅模式
    @Component
    public class Producer {
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public  void send(){
            while (true){
                System.out.println("输入消息");
                String msg = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("logs","",msg);
            }
        }
    
    }
    
    

    消费者

    消费者需要执行以下操作:

    1. 定义随机队列(随机命名,非持久,排他,自动删除)
    2. 定义交换机(可以省略, 已在主程序中定义)
    3. 将队列绑定到交换机

    spring boot 通过注解完成以上操作:

    @RabbitListener(bindings = @QueueBinding( //这里进行绑定设置
    	value = @Queue, //这里定义随机队列,默认属性: 随机命名,非持久,排他,自动删除
    	exchange = @Exchange(name = "logs", declare = "false") //指定 logs 交换机,因为主程序中已经定义,这里不进行定义
    ))
    
    package cn.tedu.rabbitmq.m3;
    
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Consumer {
        
        //定义两个消费者
        //交换机 随机队列 绑定
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "logs",declare = "false"),
                value =@Queue //若无参数 则为随机队列 非持久 独占 自动删除
    
        ))
        public void receiver1(String s){
    
            System.out.println("消费者1收到"+s);
        }
    
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "logs",declare = "false"),
                value =@Queue //若无参数 则为随机队列 非持久 独占 自动删除
    
        ))
        public void receiver2(String s){
    
            System.out.println("消费者2收到"+s);
        }
    }
    
    

    测试类

    package cn.tedu.rabbitmq.m3;
    
    import cn.tedu.rabbitmq.m3.Producer;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    @SpringBootTest
    public class Test3 {
    
        @Autowired
        Producer producer;
        @Test
        public  void  test1(){
            producer.send();
        }
    
    }
    

    路由模式

    与发布和订阅模式代码类似, 只是做以下三点调整:

    1. 使用 direct 交换机
    2. 队列和交换机绑定时, 设置绑定键
    3. 发送消息时, 指定路由键

    主程序

    主程序中使用 DirectExcnahge 对象封装交换机信息, spring boot 自动配置类会自动发现这个对象, 并在 RabbitMQ 服务器上定义这个交换机.

    package cn.tedu.rabbitmq.m4;
    
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    //路由模式
    @SpringBootApplication
    public class Main {
        public static void main(String[] args) {
            SpringApplication.run(Main.class,args);
        }
    
    
        //定义一个交换机
        @Bean
        public DirectExchange directLogsExchange(){
    
            return  new DirectExchange("direct_logs",false,false);//非持久 不自动删除
    
        }
    }
    
    

    生产者

    生产者向指定的交换机发送消息, 并指定路由键.

    package cn.tedu.rabbitmq.m4;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Scanner;
    
    //发布订阅模式
    @Component
    public class Producer {
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public  void send(){
            while (true){
                System.out.println("输入消息");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键");
                String key = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("direct_logs",key,msg);
            }
        }
    
    }
    
    

    消费者

    消费者通过注解来定义随机队列, 绑定到交换机, 并指定绑定键:

    @RabbitListener(bindings = @QueueBinding( // 这里做绑定设置
    	value = @Queue, // 定义队列, 随机命名,非持久,排他,自动删除
    	exchange = @Exchange(name = "direct_logs", declare = "false"), // 指定绑定的交换机,主程序中已经定义过队列,这里不进行定义
    	key = {"error","info","warning"} // 设置绑定键
    ))
    
    
    package cn.tedu.rabbitmq.m4;
    
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Consumer {
        //交换机 随机队列  使用绑定键绑定
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "direct_logs",declare = "false"),
                value =@Queue, //若无参数 则为随机队列 非持久 独占 自动删除
                key = {"error"}
        ))
        public void receiver1(String s){
    
            System.out.println("消费者1收到"+s);
        }
    
        //交换机 随机队列  使用绑定键绑定
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "direct_logs",declare = "false"),
                value =@Queue, //若无参数 则为随机队列 非持久 独占 自动删除
                key = {"error","info","warning"}
        ))
        public void receiver2(String s){
    
            System.out.println("消费者2收到"+s);
        }
    }
    
    

    测试类

    package cn.tedu.rabbitmq.m4;
    
    import cn.tedu.rabbitmq.m4.Producer;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    @SpringBootTest
    public class Test4 {
    
        @Autowired
        Producer producer;
        @Test
        public  void  test1(){
    
            producer.send();
        }
    }
    

    主题模式

    主题模式不过是具有特殊规则的路由模式, 代码与路由模式基本相同, 只做如下调整:

    1. 使用 topic 交换机
    2. 使用特殊的绑定键和路由键规则

    主程序

    package cn.tedu.rabbitmq.m5;
    
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    
    //路由模式
    @SpringBootApplication
    public class Main {
        public static void main(String[] args) {
            SpringApplication.run(Main.class,args);
        }
    
    
        //定义一个交换机
        @Bean
        public TopicExchange directLogsExchange(){
    
            return  new TopicExchange("topic_logs",false,false);//非持久 不自动删除
    
        }
    }
    
    

    生产者

    package cn.tedu.rabbitmq.m5;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Scanner;
    
    //发布订阅模式
    @Component
    public class Producer {
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public  void send(){
            while (true){
                System.out.println("输入消息");
                String msg = new Scanner(System.in).nextLine();
                System.out.println("输入路由键");
                String key = new Scanner(System.in).nextLine();
                amqpTemplate.convertAndSend("topic_logs",key,msg);
            }
        }
    
    }
    
    

    消费者

    package cn.tedu.rabbitmq.m5;
    
    import org.springframework.amqp.rabbit.annotation.Exchange;
    import org.springframework.amqp.rabbit.annotation.Queue;
    import org.springframework.amqp.rabbit.annotation.QueueBinding;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class Consumer {
        //交换机 随机队列  使用绑定键绑定
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "topic_logs",declare = "false"),
                value =@Queue, //若无参数 则为随机队列 非持久 独占 自动删除
                key = {"*.orange.*"}
        ))
        public void receiver1(String s){
    
            System.out.println("消费者1收到"+s);
        }
    
        //交换机 随机队列  使用绑定键绑定
        @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(name = "topic_logs",declare = "false"),
                value =@Queue, //若无参数 则为随机队列 非持久 独占 自动删除
                key = {"*.*.rabbit","lazy.#"}
        ))
        public void receiver2(String s){
    
            System.out.println("消费者2收到"+s);
        }
    }
    
    

    测试

    package cn.tedu.rabbitmq.m5;
    
    import cn.tedu.rabbitmq.m5.Producer;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    @SpringBootTest
    public class Test5 {
    
        @Autowired
        Producer producer;
        @Test
        public  void  test1(){
    
            producer.send();
        }
    }
    

    RPC异步调用

    主程序

    主程序中定义两个队列

    • 发送调用信息的队列: rpc_queue
    • 返回结果的队列: 随机命名
    package cn.tedu.m6;
    
    import java.util.UUID;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.annotation.Bean;
    
    @SpringBootApplication
    public class Main {
    
    	public static void main(String[] args) {
    		SpringApplication.run(Main.class, args);
    	}
    	@Bean
    	public Queue sendQueue() {
    		return new Queue("rpc_queue",false);
    	}
    	@Bean
    	public Queue rndQueue() {
    		return new Queue(UUID.randomUUID().toString(), false);
    	}
    }
    
    

    服务端

    rpc_queue接收调用数据, 执行运算求斐波那契数,并返回计算结果.
    @Rabbitlistener注解对于具有返回值的方法:

    • 会自动获取 replyTo 属性
    • 自动获取 correlationId 属性
    • replyTo 属性指定的队列发送计算结果, 并携带 correlationId 属性
    package cn.tedu.m6;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RpcServer {
    	@RabbitListener(queues = "rpc_queue")
    	public long getFbnq(int n) {
    		return f(n);
    	}
    
    	private long f(int n) {
    		if (n==1 || n==2) {
    			return 1;
    		}
    		return f(n-1) + f(n-2);
    	}
    }
    
    

    客户端

    使用 SPEL 表达式获取随机队列名: "#{rndQueue.name}"

    发送调用数据时, 携带随机队列名和correlationId

    从随机队列接收调用结果, 并获取correlationId

    package cn.tedu.m6;
    
    import java.util.UUID;
    
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RpcClient {
    	@Autowired
    	AmqpTemplate t;
    	
    	@Value("#{rndQueue.name}")
    	String rndQueue;
    	
    	public void send(int n) {
    		// 发送调用信息时, 通过前置消息处理器, 对消息属性进行设置, 添加返回队列名和关联id
    		t.convertAndSend("rpc_queue", (Object)n, new MessagePostProcessor() {
    			@Override
    			public Message postProcessMessage(Message message) throws AmqpException {
    				MessageProperties p = message.getMessageProperties();
    				p.setReplyTo(rndQueue);
    				p.setCorrelationId(UUID.randomUUID().toString());
    				return message;
    			}
    		});
    	}
    	
    	//从随机队列接收计算结果
    	@RabbitListener(queues = "#{rndQueue.name}")
    	public void receive(long r, @Header(name=AmqpHeaders.CORRELATION_ID) String correlationId) {
    		System.out.println("
    
    "+correlationId+" - 收到: "+r);
    	}
    	
    }
    
    

    测试类

    package cn.tedu.m6;
    
    import java.util.Scanner;
    
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    @SpringBootTest
    class TopicTests {
    	@Autowired
    	RpcClient client;
    
    	@Test
    	void test1() throws Exception {
    		while (true) {
    			System.out.print("求第几个斐波那契数: ");
    			int n = new Scanner(System.in).nextInt();
    			client.send(n);
    		}
    	}
    
    }
    
    
  • 相关阅读:
    WAVECOM CDMA Modem 发送短信
    【转】关于正则表达式匹配任意字符(包括换行符)的写法
    MS2000 差异备份 还原
    推荐一款非常适用的弹框 phpcms v9都用的这个!!!!
    正则截取内容
    javascript面向对象编程实现
    一次 全部删除MSSQL数据库用户表,存储过程
    【转】 jQuery图片预加载+等比例缩放
    多种多样的Loading特效
    关于图片轮播的几种思路
  • 原文地址:https://www.cnblogs.com/liqbk/p/13592147.html
Copyright © 2011-2022 走看看