Nicksxs's Blog

What hurts more, the pain of hard work or the pain of regret?

0%

其实这个表示有点不太对,应该是 Druid 动态切换数据源的方法,只是应用在了 springboot 框架中,准备代码准备了半天,之前在一次数据库迁移中使用了,发现 Druid 还是很强大的,用来做动态数据源切换很方便。

首先这里的场景跟我原来用的有点点区别,在项目中使用的是通过配置中心控制数据源切换,统一切换,而这里的例子多加了个可以根据接口注解配置

第一部分是最核心的,如何基于 Spring JDBC 和 Druid 来实现数据源切换,是继承了org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource 这个类,他的determineCurrentLookupKey方法会被调用来获得用来决定选择那个数据源的对象,也就是 lookupKey,也可以通过这个类看到就是通过这个 lookupKey 来路由找到数据源。

public class DynamicDataSource extends AbstractRoutingDataSource {

    @Override
    protected Object determineCurrentLookupKey() {
        if (DatabaseContextHolder.getDatabaseType() != null) {
            return DatabaseContextHolder.getDatabaseType().getName();
        }
        return DatabaseType.MASTER1.getName();
    }
}

而如何使用这个 lookupKey 呢,就涉及到我们的 DataSource 配置了,原来就是我们可以直接通过spring 的 jdbc 配置数据源,像这样

现在我们要使用 Druid 作为数据源了,然后配置 DynamicDataSource的参数,通过 key 来选择对应的 DataSource,也就是下面配的 master1 和 master2

<bean id="master1" class="com.alibaba.druid.pool.DruidDataSource" init-method="init"
          destroy-method="close"
          p:driverClassName="com.mysql.cj.jdbc.Driver"
          p:url="${master1.demo.datasource.url}"
          p:username="${master1.demo.datasource.username}"
          p:password="${master1.demo.datasource.password}"
          p:initialSize="5"
          p:minIdle="1"
          p:maxActive="10"
          p:maxWait="60000"
          p:timeBetweenEvictionRunsMillis="60000"
          p:minEvictableIdleTimeMillis="300000"
          p:validationQuery="SELECT 'x'"
          p:testWhileIdle="true"
          p:testOnBorrow="false"
          p:testOnReturn="false"
          p:poolPreparedStatements="false"
          p:maxPoolPreparedStatementPerConnectionSize="20"
          p:connectionProperties="config.decrypt=true"
          p:filters="stat,config"/>

    <bean id="master2" class="com.alibaba.druid.pool.DruidDataSource" init-method="init"
          destroy-method="close"
          p:driverClassName="com.mysql.cj.jdbc.Driver"
          p:url="${master2.demo.datasource.url}"
          p:username="${master2.demo.datasource.username}"
          p:password="${master2.demo.datasource.password}"
          p:initialSize="5"
          p:minIdle="1"
          p:maxActive="10"
          p:maxWait="60000"
          p:timeBetweenEvictionRunsMillis="60000"
          p:minEvictableIdleTimeMillis="300000"
          p:validationQuery="SELECT 'x'"
          p:testWhileIdle="true"
          p:testOnBorrow="false"
          p:testOnReturn="false"
          p:poolPreparedStatements="false"
          p:maxPoolPreparedStatementPerConnectionSize="20"
          p:connectionProperties="config.decrypt=true"
          p:filters="stat,config"/>

    <bean id="dataSource" class="com.nicksxs.springdemo.config.DynamicDataSource">
        <property name="targetDataSources">
            <map key-type="java.lang.String">
                <!-- master -->
                <entry key="master1" value-ref="master1"/>
                <!-- slave -->
                <entry key="master2" value-ref="master2"/>
            </map>
        </property>
        <property name="defaultTargetDataSource" ref="master1"/>
    </bean>

现在就要回到头上,介绍下这个DatabaseContextHolder,这里使用了 ThreadLocal 存放这个 DatabaseType,为啥要用这个是因为前面说的我们想要让接口层面去配置不同的数据源,要把持相互隔离不受影响,就使用了 ThreadLocal,关于它也可以看我前面写的一篇文章聊聊传说中的 ThreadLocal,而 DatabaseType 就是个简单的枚举

public class DatabaseContextHolder {
    public static final ThreadLocal<DatabaseType> databaseTypeThreadLocal = new ThreadLocal<>();

    public static DatabaseType getDatabaseType() {
        return databaseTypeThreadLocal.get();
    }

    public static void putDatabaseType(DatabaseType databaseType) {
        databaseTypeThreadLocal.set(databaseType);
    }

    public static void clearDatabaseType() {
        databaseTypeThreadLocal.remove();
    }
}
public enum DatabaseType {
    MASTER1("master1", "1"),
    MASTER2("master2", "2");

    private final String name;
    private final String value;

    DatabaseType(String name, String value) {
        this.name = name;
        this.value = value;
    }

    public String getName() {
        return name;
    }

    public String getValue() {
        return value;
    }

    public static DatabaseType getDatabaseType(String name) {
        if (MASTER2.name.equals(name)) {
            return MASTER2;
        }
        return MASTER1;
    }
}

这边可以看到就是通过动态地通过putDatabaseType设置lookupKey来进行数据源切换,要通过接口注解配置来进行设置的话,我们就需要一个注解

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface DataSource {
    String value();
}

这个注解可以配置在我的接口方法上,比如这样

public interface StudentService {

    @DataSource("master1")
    public Student queryOne();

    @DataSource("master2")
    public Student queryAnother();

}

通过切面来进行数据源的设置

@Aspect
@Component
@Order(-1)
public class DataSourceAspect {

    @Pointcut("execution(* com.nicksxs.springdemo.service..*.*(..))")
    public void pointCut() {

    }


    @Before("pointCut()")
    public void before(JoinPoint point)
    {
        Object target = point.getTarget();
        System.out.println(target.toString());
        String method = point.getSignature().getName();
        System.out.println(method);
        Class<?>[] classz = target.getClass().getInterfaces();
        Class<?>[] parameterTypes = ((MethodSignature) point.getSignature())
                .getMethod().getParameterTypes();
        try {
            Method m = classz[0].getMethod(method, parameterTypes);
            System.out.println("method"+ m.getName());
            if (m.isAnnotationPresent(DataSource.class)) {
                DataSource data = m.getAnnotation(DataSource.class);
                System.out.println("dataSource:"+data.value());
                DatabaseContextHolder.putDatabaseType(DatabaseType.getDatabaseType(data.value()));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @After("pointCut()")
    public void after() {
				DatabaseContextHolder.clearDatabaseType();
    }
}

通过接口判断是否带有注解跟是注解的值,DatabaseType 的配置不太好,不过先忽略了,然后在切点后进行清理

这是我 master1 的数据,

master2 的数据

然后跑一下简单的 demo,

@Override
public void run(String...args) {
	LOGGER.info("run here");
	System.out.println(studentService.queryOne());
	System.out.println(studentService.queryAnother());

}

看一下运行结果

其实这个方法应用场景不止可以用来迁移数据库,还能实现精细化的读写数据源分离之类的,算是做个简单记录和分享。

这个话题是由一次组内同学分享引出来的,首先在 springboot 2.x 开始默认使用了 cglib 作为 aop 的实现,这里也稍微讲一下,在一个 1.x 的老项目里,可以看到AopAutoConfiguration 是这样的

@Configuration
@ConditionalOnClass({ EnableAspectJAutoProxy.class, Aspect.class, Advice.class })
@ConditionalOnProperty(prefix = "spring.aop", name = "auto", havingValue = "true", matchIfMissing = true)
public class AopAutoConfiguration {

	@Configuration
	@EnableAspectJAutoProxy(proxyTargetClass = false)
	@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false", matchIfMissing = true)
	public static class JdkDynamicAutoProxyConfiguration {
	}

	@Configuration
	@EnableAspectJAutoProxy(proxyTargetClass = true)
	@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true", matchIfMissing = false)
	public static class CglibAutoProxyConfiguration {
	}

}

而在 2.x 中变成了这样

@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(prefix = "spring.aop", name = "auto", havingValue = "true", matchIfMissing = true)
public class AopAutoConfiguration {

	@Configuration(proxyBeanMethods = false)
	@ConditionalOnClass(Advice.class)
	static class AspectJAutoProxyingConfiguration {

		@Configuration(proxyBeanMethods = false)
		@EnableAspectJAutoProxy(proxyTargetClass = false)
		@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "false")
		static class JdkDynamicAutoProxyConfiguration {

		}

		@Configuration(proxyBeanMethods = false)
		@EnableAspectJAutoProxy(proxyTargetClass = true)
		@ConditionalOnProperty(prefix = "spring.aop", name = "proxy-target-class", havingValue = "true",
				matchIfMissing = true)
		static class CglibAutoProxyConfiguration {

		}

	}

为何会加载 AopAutoConfiguration 在前面的文章聊聊 SpringBoot 自动装配里已经介绍过,有兴趣的可以看下,可以发现 springboot 在 2.x 版本开始使用 cglib 作为默认的动态代理实现。

然后就是出现的问题了,代码是这样的,一个简单的基于 springboot 的带有数据库的插入,对插入代码加了事务注解,

@Mapper
public interface StudentMapper {
		// 就是插入一条数据
    @Insert("insert into student(name, age)" + "values ('nick', '18')")
    public Long insert();
}

@Component
public class StudentManager {

    @Resource
    private StudentMapper studentMapper;
    
    public Long createStudent() {
        return studentMapper.insert();
    }
}

@Component
public class StudentServiceImpl implements StudentService {

    @Resource
    private StudentManager studentManager;

    // 自己引用
    @Resource
    private StudentServiceImpl studentService;

    @Override
    @Transactional
    public Long createStudent() {
        Long id = studentManager.createStudent();
        Long id2 = studentService.createStudent2();
        return 1L;
    }

    @Transactional
    private Long createStudent2() {
//        Integer t = Integer.valueOf("aaa");
        return studentManager.createStudent();
    }
}

第一个公有方法 createStudent 首先调用了 manager 层的创建方法,然后再通过引入的 studentService 调用了createStudent2,我们先跑一下看看会出现啥情况,果不其然报错了,正是这个报错让我纠结了很久

EdR7oB

报了个空指针,而且是在 createStudent2 已经被调用到了,在它的内部,报的 studentManager 是 null,首先 cglib 作为动态代理它是通过继承的方式来实现的,相当于是会在调用目标对象的代理方法时调用 cglib 生成的子类,具体的代理切面逻辑在子类实现,然后在调用目标对象的目标方法,但是继承的方式对于 final 和私有方法其实是没法进行代理的,因为没法继承,所以我最开始的想法是应该通过 studentService 调用 createStudent2 的时候就报错了,也就是不会进入这个方法内部,后面才发现犯了个特别二的错误,继承的方式去调用父类的私有方法,对于 Java 来说是可以调用到的,父类的私有方法并不由子类的InstanceKlass维护,只能通过子类的InstanceKlass找到Java类对应的_super,这样间接地访问。也就是说子类其实是可以访问的,那为啥访问了会报空指针呢,这里报的是studentManager 是空的,可以往依赖注入方面去想,如果忽略依赖注入,我这个studentManager 的确是 null,那是不是就没有被依赖注入呢,但是为啥前面那个可以呢

这个问题着实查了很久,不废话来看代码

@Override
		protected Object invokeJoinpoint() throws Throwable {
			if (this.methodProxy != null) {
        // 这里的 target 就是被代理的 bean
				return this.methodProxy.invoke(this.target, this.arguments);
			}
			else {
				return super.invokeJoinpoint();
			}
		}

这个是org.springframework.aop.framework.CglibAopProxy.CglibMethodInvocation的代码,其实它在这里不是直接调用 super 也就是父类的方法,而是通过 methodProxy 调用 target 目标对象的方法,也就是原始的 studentService bean 的方法,这样子 spring 管理的已经做好依赖注入的 bean 就能正常起作用,否则就会出现上面的问题,因为 cglib 其实是通过继承来实现,通过将调用转移到子类上加入代理逻辑,我们在简单使用的时候会直接 invokeSuper() 调用父类的方法,但是在这里 spring 的场景里需要去支持 spring 的功能逻辑,所以上面的问题就可以开始来解释了,因为 createStudent 是公共方法,cglib 可以对其进行继承代理,但是在执行逻辑的时候其实是通过调用目标对象,也就是 spring 管理的被代理的目标对象的 bean 调用的 createStudent,而对于下面的 createStudent2 方法因为是私有方法,不会走代理逻辑,也就不会有调用回目标对象的逻辑,只是通过继承关系,在子类中没有这个方法,所以会通过子类的InstanceKlass找到这个类对应的_super,然后调用父类的这个私有方法,这里要搞清楚一个点,从这个代理类直接找到其父类然后调用这个私有方法,这个类是由 cglib 生成的,不是被 spring 管理起来经过依赖注入的 bean,所以是没有 studentManager 这个依赖的,也就出现了前面的问题

而在前面提到的cglib通过methodProxy调用到目标对象,目标对象是在什么时候设置的呢,其实是在bean的生命周期中,org.springframework.beans.factory.config.BeanPostProcessor#postProcessAfterInitialization这个接口的在bean的初始化过程中,会调用实现了这个接口的方法,

@Override
public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {
	if (bean != null) {
		Object cacheKey = getCacheKey(bean.getClass(), beanName);
		if (this.earlyProxyReferences.remove(cacheKey) != bean) {
			return wrapIfNecessary(bean, beanName, cacheKey);
		}
	}
	return bean;
}

具体的逻辑在 org.springframework.aop.framework.autoproxy.AbstractAutoProxyCreator#wrapIfNecessary这个方法里

protected Object getCacheKey(Class<?> beanClass, @Nullable String beanName) {
		if (StringUtils.hasLength(beanName)) {
			return (FactoryBean.class.isAssignableFrom(beanClass) ?
					BeanFactory.FACTORY_BEAN_PREFIX + beanName : beanName);
		}
		else {
			return beanClass;
		}
	}

	/**
	 * Wrap the given bean if necessary, i.e. if it is eligible for being proxied.
	 * @param bean the raw bean instance
	 * @param beanName the name of the bean
	 * @param cacheKey the cache key for metadata access
	 * @return a proxy wrapping the bean, or the raw bean instance as-is
	 */
	protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
		if (StringUtils.hasLength(beanName) && this.targetSourcedBeans.contains(beanName)) {
			return bean;
		}
		if (Boolean.FALSE.equals(this.advisedBeans.get(cacheKey))) {
			return bean;
		}
		if (isInfrastructureClass(bean.getClass()) || shouldSkip(bean.getClass(), beanName)) {
			this.advisedBeans.put(cacheKey, Boolean.FALSE);
			return bean;
		}

		// Create proxy if we have advice.
		Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null);
		if (specificInterceptors != DO_NOT_PROXY) {
			this.advisedBeans.put(cacheKey, Boolean.TRUE);
			Object proxy = createProxy(
					bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean));
			this.proxyTypes.put(cacheKey, proxy.getClass());
			return proxy;
		}

		this.advisedBeans.put(cacheKey, Boolean.FALSE);
		return bean;
	}

然后在 org.springframework.aop.framework.autoproxy.AbstractAutoProxyCreator#createProxy 中创建了代理类

CommitLog 结构

CommitLog 是 rocketmq 的服务端,也就是 broker 存储消息的的文件,跟 kafka 一样,也是顺序写入,当然消息是变长的,生成的规则是每个文件的默认1G =1024 * 1024 * 1024,commitlog的文件名fileName,名字长度为20位,左边补零,剩余为起始偏移量;比如00000000000000000000代表了第一个文件,起始偏移量为0,文件大小为1G=1 073 741 824Byte;当这个文件满了,第二个文件名字为00000000001073741824,起始偏移量为1073741824, 消息存储的时候会顺序写入文件,当文件满了则写入下一个文件,代码中的定义

// CommitLog file size,default is 1G
private int mapedFileSizeCommitLog = 1024 * 1024 * 1024;

kLahwW

本地跑个 demo 验证下,也是这样,这里奇妙有几个比较巧妙的点(个人观点),首先文件就刚好是 1G,并且按照大小偏移量去生成下一个文件,这样获取消息的时候按大小算一下就知道在哪个文件里了,

代码中写入 CommitLog 的逻辑可以从这开始看

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
        // Set the storage time
        msg.setStoreTimestamp(System.currentTimeMillis());
        // Set the message body BODY CRC (consider the most appropriate setting
        // on the client)
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        // Back to Results
        AppendMessageResult result = null;

        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

        String topic = msg.getTopic();
        int queueId = msg.getQueueId();

        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            // Delay Delivery
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }

                topic = ScheduleMessageService.SCHEDULE_TOPIC;
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                // Backup real topic, queueId
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
                msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }

        long eclipseTimeInLock = 0;
        MappedFile unlockMappedFile = null;
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            this.beginTimeInLock = beginLockTimestamp;

            // Here settings are stored timestamp, in order to ensure an orderly
            // global
            msg.setStoreTimestamp(beginLockTimestamp);

            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
                log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
            }

            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {
                case PUT_OK:
                    break;
                case END_OF_FILE:
                    unlockMappedFile = mappedFile;
                    // Create a new file, re-write the message
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                    if (null == mappedFile) {
                        // XXX: warn and notify me
                        log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                        beginTimeInLock = 0;
                        return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                    }
                    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                    break;
                case MESSAGE_SIZE_EXCEEDED:
                case PROPERTIES_SIZE_EXCEEDED:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
                case UNKNOWN_ERROR:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
                default:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            }

            eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            beginTimeInLock = 0;
        } finally {
            putMessageLock.unlock();
        }

        if (eclipseTimeInLock > 500) {
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, msg.getBody().length, result);
        }

        if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
            this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
        }

        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

        // Statistics
        storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
        storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());

        handleDiskFlush(result, putMessageResult, msg);
        handleHA(result, putMessageResult, msg);

        return putMessageResult;
    }

前面也看到在CommitLog 目录下是有大小为 1G 的文件组成,在实现逻辑中,其实是通过 org.apache.rocketmq.store.MappedFileQueue ,内部是存的一个MappedFile的队列,对于写入的场景每次都是通过org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile() 获取最后一个文件,如果还没有创建,或者最后这个文件已经满了,那就调用 org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile(long)

public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
        long createOffset = -1;
  			// 调用前面的方法,只是从 mappedFileQueue 获取最后一个
        MappedFile mappedFileLast = getLastMappedFile();

        // 如果为空,计算下创建的偏移量
        if (mappedFileLast == null) {
            createOffset = startOffset - (startOffset % this.mappedFileSize);
        }
  
				// 如果不为空,但是当前的文件写满了
        if (mappedFileLast != null && mappedFileLast.isFull()) {
            // 前一个的偏移量加上单个文件的偏移量,也就是 1G
            createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
        }

        if (createOffset != -1 && needCreate) {
            // 根据 createOffset 转换成文件名进行创建
            String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
            String nextNextFilePath = this.storePath + File.separator
                + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
            MappedFile mappedFile = null;

          	// 这里如果allocateMappedFileService 存在,就提交请求
            if (this.allocateMappedFileService != null) {
                mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
                    nextNextFilePath, this.mappedFileSize);
            } else {
                try {
                  // 否则就直接创建
                    mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
                } catch (IOException e) {
                    log.error("create mappedFile exception", e);
                }
            }

            if (mappedFile != null) {
                if (this.mappedFiles.isEmpty()) {
                    mappedFile.setFirstCreateInQueue(true);
                }
                this.mappedFiles.add(mappedFile);
            }

            return mappedFile;
        }

        return mappedFileLast;
    }

首先看下直接创建的,

public MappedFile(final String fileName, final int fileSize) throws IOException {
        init(fileName, fileSize);
    }
private void init(final String fileName, final int fileSize) throws IOException {
        this.fileName = fileName;
        this.fileSize = fileSize;
        this.file = new File(fileName);
        this.fileFromOffset = Long.parseLong(this.file.getName());
        boolean ok = false;

        ensureDirOK(this.file.getParent());

        try {
          // 通过 RandomAccessFile 创建 fileChannel
            this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
          // 做 mmap 映射
            this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
            TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
            TOTAL_MAPPED_FILES.incrementAndGet();
            ok = true;
        } catch (FileNotFoundException e) {
            log.error("create file channel " + this.fileName + " Failed. ", e);
            throw e;
        } catch (IOException e) {
            log.error("map file " + this.fileName + " Failed. ", e);
            throw e;
        } finally {
            if (!ok && this.fileChannel != null) {
                this.fileChannel.close();
            }
        }
    }

如果是提交给AllocateMappedFileService的话就用到了一些异步操作

public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
        int canSubmitRequests = 2;
        if (this.messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool()
                && BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) { //if broker is slave, don't fast fail even no buffer in pool
                canSubmitRequests = this.messageStore.getTransientStorePool().remainBufferNumbs() - this.requestQueue.size();
            }
        }
				// 将请求放在 requestTable 中
        AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
        boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
        // requestTable 使用了 concurrentHashMap,用文件名作为 key,防止并发
        if (nextPutOK) {
            // 这里判断了是否可以提交到 TransientStorePool,涉及读写分离,后面再细聊
            if (canSubmitRequests <= 0) {
                log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
                    "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                this.requestTable.remove(nextFilePath);
                return null;
            }
          // 塞到阻塞队列中
            boolean offerOK = this.requestQueue.offer(nextReq);
            if (!offerOK) {
                log.warn("never expected here, add a request to preallocate queue failed");
            }
            canSubmitRequests--;
        }

        // 这里的两个提交我猜测是为了多生成一个 CommitLog,
        AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
        boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
        if (nextNextPutOK) {
            if (canSubmitRequests <= 0) {
                log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
                    "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                this.requestTable.remove(nextNextFilePath);
            } else {
                boolean offerOK = this.requestQueue.offer(nextNextReq);
                if (!offerOK) {
                    log.warn("never expected here, add a request to preallocate queue failed");
                }
            }
        }

        if (hasException) {
            log.warn(this.getServiceName() + " service has exception. so return null");
            return null;
        }

        AllocateRequest result = this.requestTable.get(nextFilePath);
        try {
          // 这里就异步等着
            if (result != null) {
                boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
                if (!waitOK) {
                    log.warn("create mmap timeout " + result.getFilePath() + " " + result.getFileSize());
                    return null;
                } else {
                    this.requestTable.remove(nextFilePath);
                    return result.getMappedFile();
                }
            } else {
                log.error("find preallocate mmap failed, this never happen");
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
        }

        return null;
    }

而真正去执行文件操作的就是 AllocateMappedFileService的 run 方法

public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped() && this.mmapOperation()) {

        }
        log.info(this.getServiceName() + " service end");
    }
private boolean mmapOperation() {
        boolean isSuccess = false;
        AllocateRequest req = null;
        try {
          // 从阻塞队列里获取请求
            req = this.requestQueue.take();
            AllocateRequest expectedRequest = this.requestTable.get(req.getFilePath());
            if (null == expectedRequest) {
                log.warn("this mmap request expired, maybe cause timeout " + req.getFilePath() + " "
                    + req.getFileSize());
                return true;
            }
            if (expectedRequest != req) {
                log.warn("never expected here,  maybe cause timeout " + req.getFilePath() + " "
                    + req.getFileSize() + ", req:" + req + ", expectedRequest:" + expectedRequest);
                return true;
            }

            if (req.getMappedFile() == null) {
                long beginTime = System.currentTimeMillis();

                MappedFile mappedFile;
                if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                    try {
                      // 通过 transientStorePool 创建
                        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
                        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    } catch (RuntimeException e) {
                        log.warn("Use default implementation.");
                      // 默认创建
                        mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    }
                } else {
                  // 默认创建
                    mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
                }

                long eclipseTime = UtilAll.computeEclipseTimeMilliseconds(beginTime);
                if (eclipseTime > 10) {
                    int queueSize = this.requestQueue.size();
                    log.warn("create mappedFile spent time(ms) " + eclipseTime + " queue size " + queueSize
                        + " " + req.getFilePath() + " " + req.getFileSize());
                }

                // pre write mappedFile
                if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()
                    .getMapedFileSizeCommitLog()
                    &&
                    this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
                    mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
                        this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
                }

                req.setMappedFile(mappedFile);
                this.hasException = false;
                isSuccess = true;
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
            this.hasException = true;
            return false;
        } catch (IOException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
            this.hasException = true;
            if (null != req) {
                requestQueue.offer(req);
                try {
                    Thread.sleep(1);
                } catch (InterruptedException ignored) {
                }
            }
        } finally {
            if (req != null && isSuccess)
              // 通知前面等待的
                req.getCountDownLatch().countDown();
        }
        return true;
    }

这是个很大的话题了,可能会分成两部分说,第一部分就是所谓的零拷贝 ( zero-copy ),这一块其实也不新鲜,我对零拷贝的概念主要来自这篇文章,个人感觉写得非常好,在 rocketmq 中,最大的一块存储就是消息存储,也就是 CommitLog ,当然还有 ConsumeQueue 和 IndexFile,以及其他一些文件,CommitLog 的存储是以一个 1G 大小的文件作为存储单位,写完了就再建一个,那么如何提高这 1G 文件的读写效率呢,就是 mmap,传统意义的读写文件,read,write 都需要由系统调用,来回地在用户态跟内核态进行拷贝切换,

read(file, tmp_buf, len);
write(socket, tmp_buf, len);

vms95Z

如上面的图显示的,要在用户态跟内核态进行切换,数据还需要在内核缓冲跟用户缓冲之间拷贝多次,

  1. 第一步是调用 read,需要在用户态切换成内核态,DMA模块从磁盘中读取文件,并存储在内核缓冲区,相当于是第一次复制
  2. 数据从内核缓冲区被拷贝到用户缓冲区,read 调用返回,伴随着内核态又切换成用户态,完成了第二次复制
  3. 然后是write 写入,这里也会伴随着用户态跟内核态的切换,数据从用户缓冲区被复制到内核空间缓冲区,完成了第三次复制,这次有点不一样的是数据不是在内核缓冲区了,会复制到 socket buffer 中。
  4. write 系统调用返回,又切换回了用户态,然后数据由 DMA 拷贝到协议引擎。

如此就能看出其实默认的读写操作代价是非常大的,而在 rocketmq 等高性能中间件中都有使用的零拷贝技术,其中 rocketmq 使用的是 mmap

mmap

mmap基于 OS 的 mmap 的内存映射技术,通过MMU 映射文件,将文件直接映射到用户态的内存地址,使得对文件的操作不再是 write/read,而转化为直接对内存地址的操作,使随机读写文件和读写内存相似的速度。

mmap 把文件映射到用户空间里的虚拟内存,省去了从内核缓冲区复制到用户空间的过程,文件中的位置在虚拟内存中有了对应的地址,可以像操作内存一样操作这个文件,这样的文件读写文件方式少了数据从内核缓存到用户空间的拷贝,效率很高。

tmp_buf = mmap(file, len);
write(socket, tmp_buf, len);

I68mFx

第一步:mmap系统调用使得文件内容被DMA引擎复制到内核缓冲区。然后该缓冲区与用户进程共享,在内核和用户内存空间之间不进行任何拷贝。

第二步:写系统调用使得内核将数据从原来的内核缓冲区复制到与套接字相关的内核缓冲区。

第三步:第三次拷贝发生在DMA引擎将数据从内核套接字缓冲区传递给协议引擎时。

通过使用mmap而不是read,我们将内核需要拷贝的数据量减少了一半。当大量的数据被传输时,这将有很好的效果。然而,这种改进并不是没有代价的;在使用mmap+write方法时,有一些隐藏的陷阱。例如当你对一个文件进行内存映射,然后在另一个进程截断同一文件时调用写。你的写系统调用将被总线错误信号SIGBUS打断,因为你执行了一个错误的内存访问。该信号的默认行为是杀死进程并dumpcore–这对网络服务器来说不是最理想的操作。

有两种方法可以解决这个问题。

第一种方法是为SIGBUS信号安装一个信号处理程序,然后在处理程序中简单地调用返回。通过这样做,写系统调用会返回它在被打断之前所写的字节数,并将errno设置为成功。让我指出,这将是一个糟糕的解决方案,一个治标不治本的解决方案。因为SIGBUS预示着进程出了严重的问题,所以不鼓励使用这种解决方案。

第二个解决方案涉及内核的文件租赁(在Windows中称为 “机会锁”)。这是解决这个问题的正确方法。通过在文件描述符上使用租赁,你与内核在一个特定的文件上达成租约。然后你可以向内核请求一个读/写租约。当另一个进程试图截断你正在传输的文件时,内核会向你发送一个实时信号,即RT_SIGNAL_LEASE信号。它告诉你内核即将终止你对该文件的写或读租约。在你的程序访问一个无效的地址和被SIGBUS信号杀死之前,你的写调用会被打断了。写入调用的返回值是中断前写入的字节数,errno将被设置为成功。下面是一些示例代码,显示了如何从内核中获得租约。

if(fcntl(fd, F_SETSIG, RT_SIGNAL_LEASE) == -1) {
    perror("kernel lease set signal");
    return -1;
}
/* l_type can be F_RDLCK F_WRLCK */
if(fcntl(fd, F_SETLEASE, l_type)){
    perror("kernel lease set type");
    return -1;
}

rocketmq 里有一种比较特殊的用法,就是顺序消息,比如订单的生命周期里,在创建,支付,签收等状态轮转中,会发出来对应的消息,这里面就比较需要去保证他们的顺序,当然在处理的业务代码也可以做对应的处理,结合消息重投,但是如果这里消息就能保证顺序性了,那么业务代码就能更好的关注业务代码的处理。

首先有一种情况是全局的有序,比如对于一个 topic 里就得发送线程保证只有一个,内部的 queue 也只有一个,消费线程也只有一个,这样就能比较容易的保证全局顺序性了,但是这里的问题就是完全限制了性能,不是很现实,在真实场景里很多都是比如对于同一个订单,需要去保证状态的轮转是按照预期的顺序来,而不必要全局的有序性。

对于这类的有序性,需要在发送和接收方都有对应的处理,在发送消息中,需要去指定 selector,即MessageQueueSelector,能够以固定的方式是分配到对应的 MessageQueue

比如像 RocketMQ 中的示例

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long id = (Long) arg;  //message queue is selected by #salesOrderID
                    long index = id % mqs.size();
                    return mqs.get((int) index);
                }
            }, orderList.get(i).getOrderId());

而在消费侧有几个点比较重要,首先我们要保证一个 MessageQueue只被一个消费者消费,消费队列存在broker端,要保证 MessageQueue 只被一个消费者消费,那么消费者在进行消息拉取消费时就必须向mq服务器申请队列锁,消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中。

List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        for (MessageQueue mq : mqSet) {
            if (!this.processQueueTable.containsKey(mq)) {
              // 判断是否顺序,如果是顺序消费的,则需要加锁
                if (isOrder && !this.lock(mq)) {
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }

                this.removeDirtyOffset(mq);
                ProcessQueue pq = new ProcessQueue();
                long nextOffset = this.computePullFromWhere(mq);
                if (nextOffset >= 0) {
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                        log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                    } else {
                        log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                        PullRequest pullRequest = new PullRequest();
                        pullRequest.setConsumerGroup(consumerGroup);
                        pullRequest.setNextOffset(nextOffset);
                        pullRequest.setMessageQueue(mq);
                        pullRequest.setProcessQueue(pq);
                        pullRequestList.add(pullRequest);
                        changed = true;
                    }
                } else {
                    log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
                }
            }
        }

在申请到锁之后会创建 pullRequest 进行消息拉取,消息拉取部分的代码实现在PullMessageService中,

@Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                this.pullMessage(pullRequest);
            } catch (InterruptedException ignored) {
            } catch (Exception e) {
                log.error("Pull Message Service Run Method exception", e);
            }
        }

        log.info(this.getServiceName() + " service end");
    }

消息拉取完后,需要提交到ConsumeMessageService中进行消费,顺序消费的实现为ConsumeMessageOrderlyService,提交消息进行消费的方法为ConsumeMessageOrderlyService#submitConsumeRequest,具体实现如下:

@Override
public void submitConsumeRequest(
    final List<MessageExt> msgs,
    final ProcessQueue processQueue,
    final MessageQueue messageQueue,
    final boolean dispathToConsume) {
    if (dispathToConsume) {
        ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
        this.consumeExecutor.submit(consumeRequest);
    }
}

构建了一个ConsumeRequest对象,它是个实现了 runnable 接口的类,并提交给了线程池来并行消费,看下顺序消费的ConsumeRequest的run方法实现:

@Override
        public void run() {
            if (this.processQueue.isDropped()) {
                log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                return;
            }
						// 获得 Consumer 消息队列锁,即单个线程独占
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock) {
              // (广播模式) 或者 (集群模式 && Broker消息队列锁有效)
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                    || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    final long beginTime = System.currentTimeMillis();
                  // 循环
                    for (boolean continueConsume = true; continueConsume; ) {
                        if (this.processQueue.isDropped()) {
                            log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                            break;
                        }

                      // 消息队列分布式锁未锁定,提交延迟获得锁并消费请求
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && !this.processQueue.isLocked()) {
                            log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }

                      // 消息队列分布式锁已经过期,提交延迟获得锁并消费请求
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && this.processQueue.isLockExpired()) {
                            log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }
												// 当前周期消费时间超过连续时长,默认:60s,提交延迟消费请求。默认情况下,每消费1分钟休息10ms。
                        long interval = System.currentTimeMillis() - beginTime;
                        if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
                            ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
                            break;
                        }
												// 获取消费消息。此处和并发消息请求不同,并发消息请求已经带了消费哪些消息。
                        final int consumeBatchSize =
                            ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();

                        List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);

                            ConsumeOrderlyStatus status = null;

                            ConsumeMessageContext consumeMessageContext = null;
                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext = new ConsumeMessageContext();
                                consumeMessageContext
                                    .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
                                consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
                                consumeMessageContext.setMq(messageQueue);
                                consumeMessageContext.setMsgList(msgs);
                                consumeMessageContext.setSuccess(false);
                                // init the consume context type
                                consumeMessageContext.setProps(new HashMap<String, String>());
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
                            }
														// 执行消费
                            long beginTimestamp = System.currentTimeMillis();
                            ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
                            boolean hasException = false;
                            try {
                                this.processQueue.getLockConsume().lock(); // 锁定处理队列
                                if (this.processQueue.isDropped()) {
                                    log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
                                        this.messageQueue);
                                    break;
                                }

                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                                log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
                                    RemotingHelper.exceptionSimpleDesc(e),
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                                hasException = true;
                            } finally {
                                this.processQueue.getLockConsume().unlock();  // 解锁
                            }

                            if (null == status
                                || ConsumeOrderlyStatus.ROLLBACK == status
                                || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}",
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                            }

                            long consumeRT = System.currentTimeMillis() - beginTimestamp;
                            if (null == status) {
                                if (hasException) {
                                    returnType = ConsumeReturnType.EXCEPTION;
                                } else {
                                    returnType = ConsumeReturnType.RETURNNULL;
                                }
                            } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
                                returnType = ConsumeReturnType.TIME_OUT;
                            } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                returnType = ConsumeReturnType.FAILED;
                            } else if (ConsumeOrderlyStatus.SUCCESS == status) {
                                returnType = ConsumeReturnType.SUCCESS;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
                            }

                            if (null == status) {
                                status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.setStatus(status.toString());
                                consumeMessageContext
                                    .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
                            }

                            ConsumeMessageOrderlyService.this.getConsumerStatsManager()
                                .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } else {
                            continueConsume = false;
                        }
                    }
                } else {
                    if (this.processQueue.isDropped()) {
                        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                        return;
                    }

                    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
                }
            }
        }

获取到锁对象后,使用synchronized尝试申请线程级独占锁。

如果加锁成功,同一时刻只有一个线程进行消息消费。

如果加锁失败,会延迟100ms重新尝试向broker端申请锁定messageQueue,锁定成功后重新提交消费请求

创建消息拉取任务时,消息客户端向broker端申请锁定MessageQueue,使得一个MessageQueue同一个时刻只能被一个消费客户端消费。

消息消费时,多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁,加锁成功才能进行消费,使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费。
这里其实还有很重要的一点是对processQueue的加锁,这里其实是保证了在 rebalance的过程中如果 processQueue 被分配给了另一个 consumer,但是当前已经被我这个 consumer 再消费,但是没提交,就有可能出现被两个消费者消费,所以得进行加锁保证不受 rebalance 影响。