BsonUndefined 转换问题

关于这个问题的出现，是在业务中使用存储函数时，如果某个字段值为null，存入到数据库中会出现undefined，而随后查询会抛出ConverterNotFoundException异常信息，如下所示：

1

org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [org.bson.BsonUndefined] to type [java.lang.Integer]

若要将开源项目包发布到maven中央仓库，官方指南指出需要具备如下必要条件：

• 是releases版本
• 有javadoc和sources
• 经过GPG签名
• 必要的POM信息
• 开源包的坐标，即groupId和artifactId

那么如何发布呢？这里采用最便捷通用的一种方式，使用Sonatype提供的OSSRH(Open Source Software Repository Hosting )来进行上传。

文件上传是一个很常见的功能，在业务场景中其又可分为单文件上传、分片上传、断点续传、秒传等。

一个小文件上传在一个http连接便可以很快的完成，其无需担心上传失败重新上传的问题。而一个大文件的上传则不能这样，试想一个场景：10G的文件直接上传，如果上传一方的网速很好，服务器的网络带宽很小，那么服务器的带宽全被这个上传连接占用，其他人上传文件则已没有带宽可用；如果在网速较差的环境下上传，快要上传完成的时候网络中断，又得重新上传，一定会抓狂。

这里以 ubuntu18.04 LTS 系统为例

查看系统内核版本：

1

uname -a

查看已安装内核信息，确认需要安装的相应包：

1

sudo dpkg --get-selections |grep linux

顾名思义，ThreadLocal是为线程提供私有的局部变量。它不同于其他常规的变量，需要使用自身的get和set方法来获取和设置值。ThreadLocal的典型应用是在类中被申明为静态变量，用于关联用户ID、事务ID，亦或者其他需要线程独有的属性。

对于ThreadLocal，只要该线程处于活动状态并且ThreadLocal实例是可访问的，每个线程都保留对其本地线程副本的隐式引用。如果线程消失后，其所有副本线程本地实例便会受到垃圾回收（除非其他情况，即线程外存在对这些副本的引用）。

rocketmq支持有序的发送消息，有序的消费消息，这里便来看一下如何实现？

实战方面均以RocketMQTemplate形式展现，集成方案详见《springboot中rocketmq的集成与使用》

消息的有序发送方面，我们可以直接使用syncSendOrderly（同步有序发送）和asyncSendOrderly（异步有序发送）两种类型的方法进行发送消息，他们的区别就如命名一样是同步和异步的区别。

rocketmq支持发送事务消息，即发送消息的事务性，这里便来看一下如何实现？

实战方面均以RocketMQTemplate形式展现，集成方案详见《springboot中rocketmq的集成与使用》

首先使用@RocketMQTransactionListener定义一个监听器来模拟执行本地事务和事务会查：

行为的触发时机

消费者消费偏移位置的持久化是消费客户端的行为，是在client启动的时候设定的一个定时任务，如下所示：

1
2
3
4
5
6
7
8
9
10
11

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            MQClientInstance.this.persistAllConsumerOffset();
        } catch (Exception e) {
            log.error("ScheduledTask persistAllConsumerOffset exception", e);
        }
    }
}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

rocketmq的client在启动的时候，会通过开启一个定时任务来定期刷新topic信息，这里就来看一下这个刷新的过程。

首先来看一下这个定时任务：

1
2
3
4
5
6
7
8
9
10
11

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            MQClientInstance.this.updateTopicRouteInfoFromNameServer();
        } catch (Exception e) {
            log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
        }
    }
}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

producer在rocketmq的作用是消息的生产者，consumer在rocketmq的作用是消息的消费者，它的生命周期是跟项目相关的，即是由使用者控制的。而为什么要将这两个角色的启动关闭流程放在一起剖析呢？是因为他们都是MQ的客户端，在启动和关闭的行为上，有着很多共同的地方。接下来便将会来仔细探究其启动和关闭的过程。

Producer

DefaultMQProducer

DefaultMQProducer的启动

producer在启动的时候会做一系列的内部初始化，其启动的源码如下所示：

1
2
3
4
5
6
7
8
9
10
11

public void start() throws MQClientException {
    this.setProducerGroup(withNamespace(this.producerGroup));
    this.defaultMQProducerImpl.start();
    if (null != traceDispatcher) {
        try {
            traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel());
        } catch (MQClientException e) {
            log.warn("trace dispatcher start failed ", e);
        }
    }
}