通用接口开放平台设计与实现——（45）API服务之消息处理_app_ticket

在前面几篇，我们使用职责链的设计模式，实现了API服务调用的技术框架，本篇我们基于该技术框架，实现业务接口的开发。

与通用接口平台对接，我们推荐的方式是使用消息服务，即前面说的，使用netty框架，构建WebSocket请求，建立长连接进行数据交互。
​

但作为开发平台，需要考虑到调用系统的多样性，对于不方便使用多线程和长连接的开发语言，在API服务接口中实现一个消息接口，即调用方采用轮询机制，查询是否有新消息，有则处理并确认消息，准实时进行消息通知。
​

在消息服务的相关章节中，我们对这块也提及到，对接模式有两种，一种是基于消息客户端模式，另外一种则就是今天要说的基于API服务接口调用的方式。
​

定时轮询方式，虽然在性能和实时性上不如消息服务的方式，但对接比较简单，无需消息客户端。
​

需要注意下，跟常规的定时轮询实现方式不同的是，我们是提供两个统一的接口，一个是消息查询，另外一个是消息确认，而不是对不同的消息主题各提供两个消息接口。这么设计与实现，对接系统，只需要定时轮询1个消息查询接口，查询到的消息根据消息主题进行不同处理就行了，因此性能开销非常小。
​

​

消息查询

​

先来说消息查询，具体实现如下：
​

/**
 * 消息查询处理器
 * @author wqliu
 * @date 2021-8-20 13:15
 **/
@Slf4j
public class MessageQueryHandler extends ServiceHandler<MessageQueryParameter> {


    @Override
    protected String handleBusiness(MessageQueryParameter parameter,String appCode) {
        ApiMessageLogService service= SpringUtil.getBean(ApiMessageLogService.class);
        List<ApiMessageLog> list = service.queryWaitHandleMessages(parameter.getCount(), appCode);
        String data = JSON.toJSONString(list);
        log.info("查询到的待处理消息为：{}",data);
        return data;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

其中有个辅助的业务请求参数类如下：

/**
 * 消息查询参数
 * @author wqliu
 * @date 2021-8-20 13:17
 **/
@Data
public class MessageQueryParameter {

    /**
     * 数量
     */
    @NotNull(message = "数量不能为空")
    private Integer count;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

虽然参数只有一个，但为了通用性和扩展性，我们做了封装。
​

核心的业务逻辑实现，则在服务类中

    @Override
    public List<ApiMessageLog> queryWaitHandleMessages(int count,String appCode) {
        //处理消息数量，如为0，默认设置默认数量，如超出最大数量，则置为最大数量
        if (count == 0) {
            count = DEFAULT_QUERY_MESSAGE_COUNT;
        } else if (count > MAX_QUERY_MESSAGE_COUNT) {
            count = MAX_QUERY_MESSAGE_COUNT;
        }

        List<ApiMessageLog> list = this.lambdaQuery()
                .select(ApiMessageLog::getRequestId,ApiMessageLog::getRequestTopicCode,ApiMessageLog::getRequestData,ApiMessageLog::getRequestTime)
                .eq(ApiMessageLog::getStatus, ApiMessageStatusEnum.WAIT_HANDLE.name())
                .eq(ApiMessageLog::getResponseAppCode, appCode)
                .orderByAsc(ApiMessageLog::getRequestTime)
                //只取指定数量的消息
                .last("limit " + count)
                .list();
        return list;


    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

消息确认

消息确认的实现也类似，加入了异常处理，确认过程中发送异常，会返回给前端友好的异常信息，错误编码以B01标识。
​

/**
 * 消息确认处理器
 *
 * @author wqliu
 * @date 2022-02-14
 */
@Slf4j
public class MessageConfirmHandler extends ServiceHandler<MessageConfirmParameter> {


    @Override
    protected String handleBusiness(MessageConfirmParameter parameter,String appCode) {
        ApiMessageLogService service = SpringUtil.getBean(ApiMessageLogService.class);
        try {
            service.confirm(parameter.getMessageId(), appCode);
        } catch (Exception e) {
            throw new ApiException("B01",e.getMessage());
        }

        log.info("消息确认成功");
        return null;
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

/**
 * 消息确认参数
 *
 * @author wqliu
 * @date 2022-02-14
 */
@Data
public class MessageConfirmParameter {

    /**
     * 消息标识
     */
    @NotNull(message = "消息标识不能为空")
    private String messageId;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

    @Override
    public void confirm(String requestMessageId, String appCode) {

        //获取消息日志对象
        ApiMessageLog log = getByRequestMessageId(requestMessageId);
        //判断是否有权限对本消息确认
        if(!appCode.equals(log.getResponseAppCode())){
            throw new CustomException(ApiMessageLogExceptionEnum.MESSAGE_CONFIRM_PERMISSION_ERROR);
        }

        //更新消息日志
        log.setStatus(ApiMessageStatusEnum.HANDLED.name());
        log.setResponseResult(MessageResponseResultEnum.SUCCESS.name());
        log.setResponseTime(LocalDateTime.now());
        //更新日志
        modify(log);

    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19