深入浅出boot2.0 第13章2异步消息activeMq和rabbitMq

怼烎@ 2023-02-20 06:26 25阅读 0赞

# 异步消息 #

*  短信系统
 *  引入JMS java message service，java消息服务
 *  点对点式，point to point (一个系统的消息发布到指定的另外一个系统)
 *  发布订阅 publish / subscribe（一个系统约定将消息发布到 一个主题中topic中，各个系统就能够通过订阅 这个主题，根据发送过来的信息 处理对应的业务。）
 *  发布 订阅 模式 常用。
 *  传统的 ActiveMQ 和 分布式的 kafka
    
     *  还有AMQP一些实现的 ，比较常用的 rabbitMQ

## jms实例，Active MQ ##

*  用户名 密码 都是 admin

### 引入pom ###

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

### 配置 mq ###

# ActiveMQ地址
    spring.activemq.broker-url=tcp://localhost:61616
    
    # 配置用户名和密码
    spring.activemq.user=admin
    spring.activemq.password=admin
    
    # 是否使用发布订阅模式，默认是为false，即是用的是点对点的模式
    spring.jms.pub-sub-domain=true
    
    # 默认目的地址
    spring.jms.template.default-destination=activemq.default.destination
    
    # 是否启用连接池
    spring.activemq.pool.enabled=true
    
    # 连接池最大连接数配置
    spring.activemq.pool.max-connections=50
    
    # 让activeMQ信任User类，配置信任列表
    #spring.activemq.packages.trusted=com.springboot.chapter13.pojo,java.lang
    spring.activemq.packages.trust-all=true #信任所有

*  JMS连接工厂，连接池，和 JmsTemplate
 *  JmsTemplate
 *  Spring 4.1 提供 @JmsListener 接收信息

### 具体使用的service ###

// ActiveMQ服务接口
    public interface ActiveMqService { 
        
        // 发送消息
        public void sendMsg(String message);
        
        // 接收消息
        public void receiveMsg(String message); 
        
    }
    
    
    @Service
    public class ActiveMqServiceImpl implements ActiveMqService { 
    
    	// 注入由Spring Boot自动生产的jmsTemplate
    	@Autowired
    	private JmsTemplate jmsTemplate = null;
    	
    	/*@Value("${spring.jms.template.default-destination}") private String defaultDestination = null;*/
    
    	@Override
    	public void sendMsg(String message) { 
    		System.out.println("发送消息【" + message + "】");
    		jmsTemplate.convertAndSend(message);
    		// 自定义发送地址
    		// jmsTemplate.convertAndSend("your-destination", message);
    	}
    
    	@Override
    	// 使用注解，监听地址发送过来的消息
    	@JmsListener(destination = "${spring.jms.template.default-destination}")
    	public void receiveMsg(String message) { 
    		System.out.println("接收到消息：【" + message + "】");
    	}
    
    }

*  convertAndSend 发送消息的方法
 *  convert 转换，默认为 SimpleMessageConverter
 *  如果使用 SerializerMessageConverter 或 Jackson2JsonMessageConverter，只需要配置JmsTemplate 的 setMessageConverter

### 发送实体类 ###

public class User implements Serializable { 
    	private static final long serialVersionUID = 8081849731640304905L;
    	private Long id;
    	private String userName = null;
    	private String note = null;
    
    	public User(Long id, String userName, String note) { 
    		this.id = id;
    		this.userName = userName;
    		this.note = note;
    	}
    }
    
    public interface ActiveMqUserService { 
    
    	public void sendUser(User user);
    
    	public void receiveUser(User user);
    }
    
    
    
    @Service
    public class ActiveMqUserServiceImpl implements ActiveMqUserService { 
        
        // 注入由Spring Boot自动生产的jmsTemplate
        @Autowired
        private JmsTemplate jmsTemplate = null;
        
        // 自定义地址
        private static final String myDestination = "my-destination";
    
        @Override
        public void sendUser(User user) { 
            System.out.println("发送消息【" + user + "】");
            // 使用自定义地址发送对象
            jmsTemplate.convertAndSend(myDestination, user);
        }
    
        @Override
        // 监控自定义地址
        @JmsListener(destination = myDestination)
        public void receiveUser(User user) { 
            System.out.println("接收到消息：【" + user + "】");
        }
    }

### action ###

@Controller
    @RequestMapping("/activemq")
    public class ActiveMqController { 
        // 注入服务对象
        @Autowired
        private ActiveMqService activeMqService = null;
        
        // 注入服务对象
        @Autowired
        private ActiveMqUserService activeMqUserService = null;
        
        // 测试普通消息的发送
        @ResponseBody
        @GetMapping("/msg")
        public Map<String, Object> msg(String message) { 
            activeMqService.sendMsg(message);
            return result(true, message);
        }
        
        // 测试User对象的发送
        @ResponseBody
        @GetMapping("/user")
        public Map<String, Object> sendUser(Long id,
                String userName, String note) { 
            User user = new User(id, userName, note);
            activeMqUserService.sendUser(user);
            return result(true, user);
            
        }
        
        private Map<String, Object> result(Boolean success, Object message) { 
            Map<String, Object> result = new HashMap<>();
            result.put("success", success);
            result.put("message", message);
            return result;
        }
    }
    
    
    @Data
    public class ResultDTO<T> { 
        private Integer code;
        private String msg;
        private Boolean ifsuccess = false;
        private T result;
    }

## 使用AMQP——RabbitMQ ##

*  amqp是常用的消息协议
 *  amqp是一个 提供统一消息服务的应用层 标准协议
 *  基于此协议的 客户端 与 消息中间件 可传递消息。

### 引入pom ###

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

### 配置Spring Boot ###

#RabbitMQ 配置
    #RabbitMQ 服务器地址
    spring.rabbitmq.host=localhost
    #RabbitMQ 端口
    spring.rabbitmq.port=5672
    #RabbitMQ 用户
    spring.rabbitmq.username=admin
    #RabbitMQ 密码
    spring.rabbitmq.password=123456
    
    #是否确认发送的消息已经被消费
    spring.rabbitmq.publisher-confirms=true
    
    #RabbitMQ 的消息队列名称，由它发送字符串
    rabbitmq.queue.msg=spring-boot-queue-msg
    #RabbitMQ 的消息队列名称，由它发送用户对象
    rabbitmq.queue.user=spring-boot-queue-user

*  最后两个是自定义的配置属性、
    
     *  自定义两个消息队列的名称
 *  publisher-confirms 意味着 发送消息方可监听 发送消息到消费端是否成功
    
     *  如果成功 则会 根据设置 进行回调。

### 创建消息队列 ###

导入此包：org.springframework.amqp.core;
    	// 消息队列名称
    	@Value("${rabbitmq.queue.msg}")
    	private String msgQueueName = null;
    
    	// 用户队列名称
    	@Value("${rabbitmq.queue.user}")
    	private String userQueueName = null;
    
    	@Bean
    	public Queue createQueueMsg() { 
    		// 创建字符串消息队列，boolean值代表是否持久化消息
    		return new Queue(msgQueueName, true);
    	}
    
    	@Bean
    	public Queue createQueueUser() { 
    		// 创建用户消息队列，boolean值代表是否持久化消息
    		return new Queue(userQueueName, true);
    	}

### 用service 使用MQ ###

public interface RabbitMqService { 
       // 发送字符消息
       public void sendMsg(String msg);
       
       // 发送用户消息
       public void sendUser(User user);
    }
    
    
    @Service
    public class RabbitMqServiceImpl 
           // 实现ConfirmCallback接口，这样可以回调
           implements ConfirmCallback, RabbitMqService { 
    
       @Value("${rabbitmq.queue.msg}")
       private String msgRouting = null;
       
       @Value("${rabbitmq.queue.user}")
       private String userRouting = null;
       
       // 注入由Spring Boot自动配置的RabbitTemplate
       @Autowired
       private RabbitTemplate rabbitTemplate = null;
    
       // 发送消息
       @Override
       public void sendMsg(String msg) { 
           System.out.println("发送消息: 【" + msg + "】");
           // 设置回调
           rabbitTemplate.setConfirmCallback(this);
           // 发送消息，通过msgRouting确定队列
           rabbitTemplate.convertAndSend(msgRouting, msg);
       }
    
       // 发送用户
       @Override
       public void sendUser(User user) { 
           System.out.println("发送用户消息: 【" + user + "】");
           // 设置回调
           rabbitTemplate.setConfirmCallback(this);
           rabbitTemplate.convertAndSend(userRouting, user);
       }
    
       // 回调确认方法
       @Override
       public void confirm(CorrelationData correlationData, 
               boolean ack, String cause) { 
           if (ack) { 
               System.out.println("消息成功消费");
           } else { 
               System.out.println("消息消费失败:" + cause);
           }
       }
    
    }

*  实现 confirmCallback，可以作为 MQ 的生产者的回调类
 *  setConfirmCallback(this) 设置了回调对象为 当前对象
    
     *  发送消息后，当消费者 得到消息时，它就会调用 confirm方法
 *  convertAndSend 转换 和 发送消息
    
     *  通过 SimpleMessageConvert对象转换（默认）
 *  convertAndSend(msgRouting, msg) 消息队列的名称

### MQ接收器 ###

//这是一个类
    @Component
    public class RabbitMessageReceiver { 
    	
    	// 定义监听字符串队列名称
    	@RabbitListener(queues = {  "${rabbitmq.queue.msg}" })
    	public void receiveMsg(String msg) { 
     		System.out.println("收到消息: 【" + msg + "】");
    	}
    	
        // 定义监听用户队列名称
    	@RabbitListener(queues = {  "${rabbitmq.queue.user}" })
    	public void receiveUser(User user) { 
    		System.out.println("收到用户信息【" + user + "】");
    	}
    }

### action ###

@RestController
    @RequestMapping("/rabbitmq")
    public class RabbitMqController { 
        // 注入Spring Boot自定生成的对象
        @Autowired
        private RabbitMqService rabbitMqService = null;
        
        @GetMapping("/msg") // 字符串
        public Map<String, Object> msg(String message) { 
            rabbitMqService.sendMsg(message);
            return resultMap("message", message);
        }
        
        @GetMapping("/user") // 用户
        public Map<String, Object> user(Long id, String userName, String note) { 
            User user = new User(id, userName, note);
            rabbitMqService.sendUser(user);
            return resultMap("user", user);
        }
        // 结果Map
        private Map<String, Object> resultMap(String key, Object obj) { 
            Map<String, Object> result = new HashMap<>();
            result.put("success", true);
            result.put(key, obj);
            return result;
        }
    }

### docker启动mq ###

docker search rabbitmq:management
    docker pull rabbitmq:management
    
    docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 rabbitmq:managemen
    
    docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management

> –hostname：指定容器主机名称  
> –name:指定容器名称  
> \-p:将mq端口号映射到本地

*  账号密码guest，guest

### 测试 ###

发送消息: 【测试的消息内容】
    收到消息: 【测试的消息内容】   
    消息成功消费   (// 回调确认方法 )
    
    发送用户消息: 【com.hua.testj.pojo.User@1b52a6be】
    收到用户信息【com.hua.testj.pojo.User@2fa71c3b】
    消息成功消费