异步消息与异步调用

Posted 2023-02-18 阿_毅

开心一笑

【老爸斗地主把豆豆输光了，叫我给他充值。我说他不要在游戏里花钱，打发时间玩玩得了。老爸一下火了：小时候你要哪个玩具老子不给你买了，现在让你给我买点豆豆你都不肯，看来老了是指望不上你了。。。
我。。。】

新书购买

戳图购买 >>>

11.1 JMS消息介绍

11.1.1 JMS概述

JMS（Java Message Service，即Java消息服务）是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC，它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。
JMS支持两种消息发送和接收模型。一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者（Producer）发送消息到队列，消息消费者（Consumer）从队列中接收消息，队列的存在使得消息的异步传输成为可能。P2P模式图如图11-1所示。

图11-1 P2P模式图
P2P的特点是每个消息只有一个消费者 (即一旦被消费，消息就不在消息队列中)，发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响消息被发送到队列中，接收者在成功接收消息之后需向队列应答成功。
另一种称为Pub / Sub（Publish / Subscribe，即发布-订阅）模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为Topic（主题）。主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。

图11-2 Pub / Sub模式图
Pub/Sub的特点是每个消息可以有多个消费者，发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。如果你希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型。

11.2 Spring Boot集成ActiveMQ

11.2.1 ActiveMQ概述

MQ英文名MessageQueue，中文名消息队列，是一个消息的接受和转发的容器，可用于消息推送。ActiveMQ是Apache提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。

11.2.2 ActiveMQ安装

安装ActiveMQ之前，我们需要到官网（http://activemq.apache.org/activemq-5150-release.html）下载，本书使用apache-activemq-5.15.0这个版本进行讲解。ActiveMQ具体安装步骤如下：
将官网下载的安装包apache-activemq-5.15.0-bin.zip解压。
打开解压的文件夹，进入bin目录，根据电脑操作系统是32位还是64位，选择进入【win32】文件夹或者【win64】文件夹。

图11-3 ActiveMQ安装界面

双击【activemq.bat】，即可启动ActiveMQ，如图11-3所示。当看到如图11-4所示的启动信息时，代表ActiveMQ安装成功。从图中可以看出，ActiveMQ默认启动到8161端口。

图11-4 ActiveMQ启动成功界面

安装成功之后，我们在浏览器中输入http://localhost:8161/admin链接访问，第一次访问需要输入用户名admin和密码admin进行登录，登录成功之后，我们就可以看到ActiveMQ的首页。具体如图11-5所示。

图11-5 ActiveMQ首页

11.2.3 引入依赖

在Spring Boot中集成ActiveMQ，首先需要在pom.xml文件中引入所需的依赖，具体代码如下：

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

11.2.4 添加ActiveMQ配置

依赖添加完成之后，我们需要在application.properties配置文件中添加ActiveMQ配置，具体代码如下：

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.in-memory=true
spring.activemq.pool.enabled=false
spring.activemq.packages.trust-all=true
spring.activemq.packages.trust-all：

spring.activemq.packages.trust-all：ObjectMessage的使用机制是不安全的，ActiveMQ自5.12.2和5.13.0之后，强制Consumer端声明一份可信任的包列表，只有当ObjectMessage中的Object在可信任包内，才能被提取出来。改配置表示信任所有的包。

11.3 使用ActiveMQ

11.3.1 创建生产者

ActiveMQ依赖和配置开发完成之后，我们首先在数据库中建立说说表ay_mood。具体建表语句如下：

DROP TABLE IF EXISTS `ay_mood`;
CREATE TABLE `ay_mood` (
  `id` varchar(32) NOT NULL,
  `content` varchar(256) DEFAULT NULL,
  `user_id` varchar(32) DEFAULT NULL,
  `praise_num` int(11) DEFAULT NULL,
  `publish_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`),
  KEY `mood_user_id_index` (`user_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

数据库表建好之后，我们生成对应的Java Bean对象，具体代码如下：

/**
 * 描述：微信说说实体
 * @author Ay
 * @date   2017/11/28.
 */
@Entity
@Table(name = "ay_mood")
public class AyMood implements Serializable 
    //主键
    @Id
    private String id;
    //说说内容
    private String content;
    //用户id
    private String userId;
    //点赞数量
    private Integer praiseNum;
    //发表时间
    private Date publishTime;

    //省略set、get方法

AyMood实体对象开发完成之后，我们开发对应的AyMoodRepository接口，具体代码如下：

/**
 * 描述：说说repository
 * @author Ay
 * @date   2017/12/02
 */
public interface AyMoodRepository extends JpaRepository<AyMood,String> 

Repository接口开发完成之后，我们开发对应的说说服务层接口AyMoodService和相应的实现类AyMoodServiceImpl。AyMoodService具体代码如下：

/**
 * 描述：微信说说服务层
 * @author Ay
 * @date   2017/12/2.
 */
public interface AyMoodService 
    
    AyMood save(AyMood ayMood);

AyMoodService代码很简单，只有一个保存说说的方法save（），AyMoodService开发完成之后，我们实现该接口，具体代码如下：

/**
 * 描述：微信说说服务层
 * @author Ay
 * @date   2017/12/2
 */
@Service
public class AyMoodServiceImpl implements AyMoodService

    @Resource
    private AyMoodRepository ayMoodRepository;

    @Override
    public AyMood save(AyMood ayMood) 
        return ayMoodRepository.save(ayMood);
    

在实现类AyMoodServiceImpl中，我们注入AyMoodRepository接口，并调用其提供的sava（）方法，保存说说到数据库。代码开发完成之后，我们在测试类MySpringBootApplicationTests下添加测试方法：

@Resource
private AyMoodService ayMoodService;

@Test
public void testAyMood()
    AyMood ayMood = new AyMood();
    ayMood.setId("1");
    //用户阿毅id为1
    ayMood.setUserId("1");
    ayMood.setPraiseNum(0);
    //说说内容
    ayMood.setContent("这是我的第一条微信说说！！！");
    ayMood.setPublishTime(new Date());
    //往数据库保存一条数据，代码用户阿毅发表了一条说说
    AyMood mood = ayMoodService.save(ayMood);

测试用例testAyMood开发完成之后，我们允许它。运行成功之后，我们可以在数据库表ay_mood中看到一条记录，具体如图

图11-6 说说表记录
用户发表说说的类虽然都开发完成了，但是有一个问题，我们都知道微信的用户量极大，每天都有几亿的用户发表不同的说说，如果按照我们上面的做法，用户每发一条说说，后端都会单独开一个线程，将该说说的内容实时的保存到数据库中。我们都知道后端服务系统的线程数和数据库线程池中的线程数量都是固定而且宝贵的，因此将用户发表的说说实时保存到数据库中，必然造成后端服务和数据库极大的压力。所有我们使用ActiveMQ做异步消费，来抵抗用户大并发发表说说而产生的压力，提高系统整体的性能。
下面我们来开发生产者和消费者。生产者AyMoodProducer代码如下：

/**
 * 生产者
 * @author Ay
 * @date 2017/11/30
 */
@Service
public class AyMoodProducer

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void sendMessage(Destination destination, final String message) 
        jmsMessagingTemplate.convertAndSend(destination, message);
    

JmsMessagingTemplate：发消息的工具类，也可以注入JmsTemplate，JmsMessagingTemplate是对JmsTemplate进行了封装。参数destination是发送到的队列，message是待发送的消息。

11.3.2 创建消费者

生产者AyMoodProducer开发完成之后，我们来开发消费者AyMoodConsumer，具体代码如下：

/**
 * 消费者
 * @author Ay
 * @date   2017/11/30
 */
@Component
public class AyMoodConsumer

    @JmsListener(destination = "ay.queue")
    public void receiveQueue(String text) 
        System.out.println("用户发表说说【" + text + "】成功");
    

@JmsListener：使用JmsListener配置消费者监听的队列ay.queue，其中text是接收到的消息。

11.3.3 测试

生产者和消费者开发完成之后，我们在测试类MySpringBootApplicationTests下开发测试方法testAyMood（），具体代码如下：

@Resource
private AyMoodProducer ayMoodProducer;

@Test
public void testActiveMQ() 

    Destination destination = new ActiveMQQueue("ay.queue");
    ayMoodProducer.sendMessage(destination, "hello,mq!!!");

测试方法开发完成之后，运行测试方法，我们可以在控制看到打印的信息，具体如图11-7所示。同时我们可以在浏览器访问http://localhost:8161/admin/，查看队列ay.queue的消费情况，具体如图11-8所示。

图11-7 控制台打印信息 图11-8 ay.queue消费情况

生产者和消费者开发完成之后，现在我们把用户发说说改成异步消费模式。首先我们在AyMoodService类下添加异步保存接口asynSave（），具体代码如下：

/**
 * 描述：微信说说服务层
 * @author Ay
 * @date   2017/12/2.
 */
public interface AyMoodService 
    AyMood save(AyMood ayMood);
    String asynSave(AyMood ayMood);

然后我们在类AyMoodServiceImpl下实现asynSave方法，asynSave方法并不保存说说记录，而是调用AyMoodProducer类的sendMessage推送消息，完整代码如下：

/**
 * 描述：微信说说服务层
 * @author Ay
 * @date   2017/12/2
 */
@Service
public class AyMoodServiceImpl implements AyMoodService

    @Resource
    private AyMoodRepository ayMoodRepository;
    @Override
    public AyMood save(AyMood ayMood) 
        return ayMoodRepository.save(ayMood);
    
    //队列
    private static Destination destination = new ActiveMQQueue("ay.queue.asyn.save");

    @Resource
    private AyMoodProducer ayMoodProducer;
    @Override
    public String asynSave(AyMood ayMood)
        //往队列ay.queue.asyn.save推送消息，消息内容为说说实体
        ayMoodProducer.sendMessage(destination, ayMood);
        return "success";
    

其次，我们在AyMoodProducer生产者类下添加sendMessage(Destination destination, final AyMood ayMood)方法，消息内容是ayMood实体对象。AyMoodProducer生产者完整代码如下：

/**
 * 生产者
 * @author Ay
 * @date 2017/11/30
 */
@Service
public class AyMoodProducer 

    @Resource
    private JmsMessagingTemplate jmsMessagingTemplate;

    public void sendMessage(Destination destination, final String message) 
        jmsMessagingTemplate.convertAndSend(destination, message);
    

    public void sendMessage(Destination destination, final AyMood ayMood) 
        jmsMessagingTemplate.convertAndSend(destination, ayMood);
    

最后，我们修改AyMoodConsumer消费者，在receiveQueue方法中保持说说记录，完整代码如下：

/**
 * 消费者
 * @author Ay
 * @date   2017/11/30
 */
@Component
public class AyMoodConsumer 

    @JmsListener(destination = "ay.queue")
    public void receiveQueue(String text) 
        System.out.println("用户发表说说【" + text + "】成功");
    

    @Resource
    private AyMoodService ayMoodService;

    @JmsListener(destination = "ay.queue.asyn.save")
    public void receiveQueue(AyMood ayMood)
        ayMoodService.save(ayMood);
    

用户发表说说，异步保存所有代码开发完成之后，我们在测试类MySpringBootApplicationTests下添加testActiveMQAsynSave测试方法，具体代码如下：

@Test
public void testActiveMQAsynSave() 
    AyMood ayMood = new AyMood();
    ayMood.setId("2");
    ayMood.setUserId("2");
    ayMood.setPraiseNum(0);
    ayMood.setContent("这是我的第一条微信说说！！！");
    ayMood.setPublishTime(new Date());
    String msg = ayMoodService.asynSave(ayMood);
    System.out.println("异步发表说说 :" + msg);

运行测试方法testActiveMQAsynSave，成功之后，我们可以在数据库表ay_mood查询到用户id为2发表的记录，具体如图11-9所示:

图11-9 ay_mood表记录

11.4 Spring Boot异步调用

11.4.1 异步调用介绍

异步调用是相对于同步调用而言的，同步调用是指程序按预定顺序一步步执行，每一步必须等到上一步执行完成之后才能执行，而异步调用则无需等待上一步程序执行完成即可执行。在日常开发的项目中，当访问的接口较慢或者做耗时任务时，不想程序一直卡在耗时任务上，想让程序能够并行执行，我们除了可以使用多线程来并行的处理任务，也可以使用Spring Boot提供的异步处理方式@Async来处理。在Spring Boot框架中，只要提过@Async注解就能将普通的同步任务改为异步调用任务。

11.4.2 @Async使用

使用@Async注解之前，我们需要在入口类添加注解@EnableAsync开启异步调用，具体代码如下：

@SpringBootApplication
@ServletComponentScan
@ImportResource(locations="classpath:spring-mvc.xml")
@EnableAsync
public class MySpringBootApplication 

	public static void main(String[] args) 
		SpringApplication.run(MySpringBootApplication.class, args);
	

然后，我们修改AyUserServiceImpl类的findAll方法，是它能够记录方法执行花费的时间，具体代码如下：

@Override
public List<AyUser> findAll() 
    try
        System.out.println("开始做任务");
        long start = System.currentTimeMillis();
        List<AyUser> ayUserList = ayUserRepository.findAll();
        long end = System.currentTimeMillis();
        System.out.println("完成任务，耗时：" + (end - start) + "毫秒");
        return ayUserList;
    catch (Exception e)
        logger.error("method [findAll] error",e);
        return Collections.EMPTY_LIST;
    

11.4.3 测试

AyUserServiceImpl类的方法findAll（）开发完成之后，我们在MySpringBootApplicationTests测试类下开发测试方法testAsync（），该方法调用3次findAll（），并记录总共消耗的时间，由于现在是同步调用，所以代码按照顺序一步一步执行。testAsync方法具体代码如下：

@Test
public void testAsync()
    long startTime = System.currentTimeMillis();
    System.out.println("第一次查询所有用户！");
    List<AyUser> ayUserList = ayUserService.findAll();
    System.out.println("第二次查询所有用户！");
    List<AyUser> ayUserList2 = ayUserService.findAll();
    System.out.println("第三次查询所有用户！");
    List<AyUser> ayUserList3 = ayUserService.findAll();
    long endTime = System.currentTimeMillis();
    System.out.println("总共消耗：" + (endTime - startTime) + "毫秒");

测试方法testAsync（）开发完成之后，我们运行它，运行成功之后，可以在控制台看到如下的打印信息。

第一次查询所有用户！
开始做任务
完成任务，耗时：371毫秒
第二次查询所有用户！
开始做任务
完成任务，耗时：34毫秒
第三次查询所有用户！
开始做任务
完成任务，耗时：32毫秒
总共消耗：438毫秒

从打印结果可以看出，调用3次findAll（），总共消耗438毫秒。现在我们在AyUserService接口中添加异步查询方法findAsynAll（），并在AyUserServiceImpl类实现该方法，具体代码如下：

/**
 * 描述：用户服务层接口
 * @author 阿毅
 * @date   2017/10/14
 */
public interface AyUserService 

    //省略大量代码

    List<AyUser> findAll();
    //异步查询
    Future<List<AyUser>> findAsynAll();

AyUserServiceImpl类中实现findAsynAll（）方法，并在方法上添加异步调用注解@Async，具体代码如下：

@Override
@Async
public Future<List<AyUser>> findAsynAll() 
    try
        System.out.println("开始做任务");
        long start = System.currentTimeMillis();
        List<AyUser> ayUserList = ayUserRepository.findAll();
        long end = System.currentTimeMillis();
        System.out.println("完成任务，耗时：" + (end - start) + "毫秒");
        return new AsyncResult<List<AyUser>>(ayUserList) ;
    catch (Exception e)
        logger.error("method [findAll] error",e);
        return new AsyncResult<List<AyUser>>(null);
    

findAsynAll（）方法开发完成之后，我们在MySpringBootApplicationTests测试类下开发测试方法testAsync2（），具体代码如下：

@Test
public void testAsync2()throws Exception
    long startTime = System.currentTimeMillis();
    System.out.println("第一次查询所有用户！");
    Future<List<AyUser>> ayUserList = ayUserService.findAsynAll();
    System.out.println("第二次查询所有用户！");
    Future<List<AyUser>> ayUserList2 = ayUserService.findAsynAll();
    System.out.println("第三次查询所有用户！");
    Future<List<AyUser>> ayUserList3 = ayUserService.findAsynAll();
    while (true)
        if(ayUserList.isDone() && ayUserList2.isDone() && ayUserList3.isDone())
            break;
        else 
            Thread.sleep(10);
        
    
    long endTime = System.currentTimeMillis();
    System.out.println("总共消耗：" + (endTime - startTime) + "毫秒");

测试方法testAsync2（）开发完成之后，我们运行它，运行成功之后，可以在控制台看到如下的打印信息。

第一次查询所有用户！
第二次查询所有用户！
第三次查询所有用户！
开始做任务
开始做任务
开始做任务
完成任务，耗时：316毫秒
完成任务，耗时：316毫秒
完成任务，耗时：315毫秒
总共消耗：334毫秒

从上面的打印结果可以看出，testAsync2方法执行速度比testAsync方法快104毫秒（438-334）。由此说明异步调用的速度比同步调用快。

---

读书感悟

来自《足球经济学》

• 世界上没有一个足球俱乐部设有人力资源部门。
• 一般人看足球，厉害的人看足球经济学。
• 既然如此糟糕，足球生意为什么能长期存在？根本原因，在于足球球迷的本地属性。和其他商品不同，足球俱乐部球队不管多糟糕，都会有一批忠实的本地球迷。消费者不能容忍一个比过去差的产品，但球迷却能接受球队比过去差。有了球迷的关注，在市场经济条件下，自然会有人愿意投资。所以作者说：尽管各家俱乐部的经营愚蠢而无能，但足球仍是地球上最稳定的营生之一。

---

经典故事

两食人族到某公司上班，老板说：“如果你们在公司吃人，立马开除！”三个月下来相安无事，突然一天老板把他们叫到办公室骂到：“叫你们不要吃人你们还吃，明天你们不用来上班了！”两食人族离开时，一个忍不住骂到：“我们每天吃一个部门经理，什么事都没有，昨天你吃了个清洁工，今天就被发现了”。

---

大神文章

无

---

其他

如果有带给你一丝丝小快乐，就让快乐继续传递下去，欢迎鼓励，点赞、顶、欢迎留下宝贵的意见、多谢支持！