分区详解-canal.mq.partitionHash

// 共16个queue
canal.mq.partitionsNum=16
// 根据policyNo分到不同的queue，以此来保证顺序消费
canal.mq.partitionHash=ins_core_fccb_db.prpcmain:policyno,ins_core_fccb_db.prpcinsured:policyno

com.alibaba.otter.canal.connector.rocketmq.producer.CanalRocketMQProducer#send
-->
com.alibaba.otter.canal.connector.core.producer.MQMessageUtils#messagePartition

if (flatMessage.getIsDdl()) {
	partitionMessages[0] = flatMessage;
}

com.alibaba.otter.canal.connector.core.producer.MQMessageUtils#getPartitionHashColumns

partitionMessages[0] = flatMessage;

if (hashMode.tableHash) {
	int hashCode = table.hashCode();
	int pkHash = Math.abs(hashCode) % partitionsNum;
	// math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
	pkHash = Math.abs(pkHash);
	partitionMessages[pkHash] = flatMessage;
}

for (Map row : flatMessage.getData()) {
	int hashCode = 0;
	if (databaseHash) {
		hashCode = database.hashCode();
	}
	if (pkNames != null) {
		for (String pkName : pkNames) {
			String value = row.get(pkName);
			if (value == null) {
				value = "";
			}
			hashCode = hashCode ^ value.hashCode();
		}
	}

	int pkHash = Math.abs(hashCode) % partitionsNum;
	// math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
	pkHash = Math.abs(pkHash);

	FlatMessage flatMessageTmp = partitionMessages[pkHash];
	...
}