Canal订阅binlog变更并结合kafka实现消息缓冲

阿里Canal项目请先了解：canal

考虑可能binlog大批量变更，如果直接通过Canal订阅binlog变动，会造成CanalClient会瞬间爆掉。为了解决这个问题，我们可以引入kafka做一层封装，可以解决这个问题。

公司实现一套框架，拿来分享大家。感谢原作者.

1. 服务端－封装Canal订阅binlog消息并推送到kafka

binlogService server 启动端:

import java.util.concurrent.Executors

import com.today.data.transfer.UTIL._

import com.today.data.transfer.canal.CanalClient

import com.today.data.transfer.kafka.BinlogKafkaProducer

import com.today.data.transfer.util.SysEnvUtil

import com.typesafe.config.ConfigFactory

import org.slf4j.LoggerFactory

/**

  *

  * 描述: binlogService server 启动端

  *

  * @author hz.lei

  * @since 2018年03月07日 上午1:08

  */

object BinLogServer {

  val logger = LoggerFactory.getLogger(getClass)

  def main(args: Array[String]) {

    startServer()

  }

  /**

    * 以Java 环境变量模式启动

    */

  def startServer(): Unit = {

    logger.info(s"启动服务 binlogServer...")

    val producerBrokerHost = SysEnvUtil.CANAL_KAFKA_HOST

    val topic = SysEnvUtil.CANAL_KAFKA_TOPIC

    val canalServerIp = SysEnvUtil.CANAL_SERVER_IP

    val canalServerPort = SysEnvUtil.CANAL_SERVER_PORT.toInt

    val destination = SysEnvUtil.CANAL_DESTINATION

    val username = SysEnvUtil.CANAL_USERNAME

    val password = SysEnvUtil.CANAL_PASSWORD

    val kafkaProducer = new BinlogKafkaProducer(producerBrokerHost, topic)

    kafkaProducer.init()

    val canalClient = new CanalClient(canalServerIp, canalServerPort, destination, username, password);

    canalClient.registerBinlogListener(kafkaProducer)

    val executorService = Executors.newFixedThreadPool(1)

    executorService.execute(canalClient)

    logger.info("启动服务 binlogService 成功...")

  }

  def startServerWithScala(): Unit = {

    logger.info(s"启动服务 binlogServer...")

    val config = ConfigFactory.load()

    val producerBrokerHost = config.getStringProxy("kafka.producerBrokerHost")

    val topic = config.getStringProxy("kafka.topic")

    val canalServerIp = config.getStringProxy("canal.canalServerIp")

    val canalServerPort = config.getStringProxy("canal.canalServerPort").toInt

    val destination = config.getStringProxy("canal.destination")

    val username = config.getStringProxy("canal.username")

    val password = config.getStringProxy("canal.password")

    val kafkaProducer = new BinlogKafkaProducer(producerBrokerHost, topic)

    kafkaProducer.init()

    val canalClient = new CanalClient(canalServerIp, canalServerPort, destination, username, password);

    canalClient.registerBinlogListener(kafkaProducer)

    val executorService = Executors.newFixedThreadPool(1)

    executorService.execute(canalClient)

    logger.info("启动服务 binlogService 成功...")

  }

}

将收到的cannal 消息 发送到kafka:

import com.alibaba.otter.canal.protocol.CanalEntry;

import com.today.data.transfer.listener.CanalBinaryListener;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.apache.kafka.common.serialization.ByteArraySerializer;

import org.apache.kafka.common.serialization.IntegerSerializer;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.util.Properties;

/**

 * 描述: 将收到的cannal 消息 发送到kafka

 *

 * @author hz.lei

 * @date 2018年03月07日 上午12:44

 */

public class BinlogKafkaProducer implements CanalBinaryListener {

    private static Logger logger = LoggerFactory.getLogger(BinlogKafkaProducer.class);

    private String topic;

    private String host;

    protected Producer<Integer, byte[]> producer;

    public BinlogKafkaProducer(String kafkaHost, String topic) {

        this.topic = topic;

        this.host = kafkaHost;

    }

    public void init() {

        logger.info("[KafkaStringProducer] [init] " +

                ") broker-list(" + host + " )");

        Properties properties = KafkaConfigBuilder.defaultProducer().bootstrapServers(host)

                .withKeySerializer(IntegerSerializer.class)

                .withValueSerializer(ByteArraySerializer.class)

                .build();

        producer = new KafkaProducer<>(properties);

    }

    /**

     * 异步回调模式发送消息

     *

     * @param topic

     * @param message

     */

    public void send(String topic, byte[] message) {

        producer.send(new ProducerRecord<>(topic, message), (metadata, e) -> {

            if (e != null) {

                logger.error("[" + getClass().getSimpleName() + "]: 消息发送失败,cause: " + e.getMessage(), e);

            }

            logger.info("[binlog]:消息发送成功,topic:{}, offset:{}, partition:{}, time:{}",

                    metadata.topic(), metadata.offset(), metadata.partition(), metadata.timestamp());

        });

    }

    @Override

    public void onBinlog(CanalEntry.Entry entry) {

        send(topic, entry.toByteArray());

    }

}

Canal 客户端,监听处理逻辑:

import com.alibaba.otter.canal.client.CanalConnector;

import com.alibaba.otter.canal.client.CanalConnectors;

import com.alibaba.otter.canal.protocol.CanalEntry.Column;

import com.alibaba.otter.canal.protocol.CanalEntry.RowData;

import com.alibaba.otter.canal.protocol.CanalEntry.Entry;

import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;

import com.alibaba.otter.canal.protocol.CanalEntry.Header;

import com.alibaba.otter.canal.protocol.CanalEntry.EventType;

import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;

import com.alibaba.otter.canal.protocol.Message;

import com.today.data.transfer.listener.CanalBinaryListener;

import com.today.data.transfer.listener.CanalGsonListener;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

import java.text.SimpleDateFormat;

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

/**

 * 描述: Canal 客户端,监听处理 逻辑

 *

 * @author hz.lei

 * @date 2018年03月06日 下午8:21

 */

public class CanalClient implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(CanalClient.class);

    private String hostname;

    private int port;

    private String destination;

    private String username;

    private String password;

    private CanalConnector connector;

    private final static int BatchSize = 1000;

    private final static long Sleep = 1000;

    private boolean runing = false;

    private List<CanalGsonListener> gsonListeners = new ArrayList<>();

    private List<CanalBinaryListener> binaryListeners = new ArrayList<>();

    /**

     * 构造函数

     *

     * @param hostname    canal服务端的ip

     * @param port        canal服务端 port

     * @param destination canal 实例地址

     * @param username    canal用户名

     * @param password    canal密码

     */

    public CanalClient(String hostname, int port, String destination, String username, String password) {

        this.hostname = hostname;

        this.port = port;

        this.destination = destination;

        this.username = username;

        this.password = password;

        init();

    }

    public void init() {

        try {

            logger.info(new StringBuffer("[Canal实例信息 CanalClient] [start] ")

                    .append("hostname: (").append(hostname)

                    .append("), port: (").append(port)

                    .append("), destination: (").append(destination)

                    .append("), username: (").append(username)

                    .append("), password: (").append(password).append(")").toString());

            connector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname, port), destination, username, password);

            connector.connect();

            connector.subscribe(".*\\..*");

        } catch (Exception e) {

            logger.error("[CanalClient] [init] " + e.getMessage(), e);

        }

    }

    public void registerBinlogListener(CanalBinaryListener listener) {

        if (listener != null) {

            binaryListeners.add(listener);

        }

    }

    public void unregisterBinlogListener(CanalBinaryListener listener) {

        if (listener != null) {

            binaryListeners.remove(listener);

        }

    }

    @Override

    public void run() {

        logger.info("[CanalClient] [run] ");

        runing = true;

        work();

    }

    /**

     * 处理工作 work

     */

    private void work() {

        try {

            while (runing) {

                Message message = connector.getWithoutAck(BatchSize);

                long batchId = message.getId();

                int size = message.getEntries().size();

                if (batchId == -1 || size == 0) {

                    try {

                        Thread.sleep(Sleep);

                    } catch (InterruptedException e) {

                        logger.error(e.getMessage(), e);

                    }

                } else {

                    if(logger.isDebugEnabled()) {

                        logger.debug("读取binlog日志 batchId: {}, size: {}, name: {}, offsets:{}", batchId, size,

                                message.getEntries().get(0).getHeader().getLogfileName(),

                                message.getEntries().get(0).getHeader().getLogfileOffset());

                    }

                    //处理消息

                    process(message.getEntries());

                }

                // 提交确认

                connector.ack(batchId);

            }

        } catch (Exception e) {

            connector.disconnect();

            logger.error("[CanalClient] [run] " + e.getMessage(), e);

        } finally {

            reconnect();

        }

    }

    /**

     * 重连策略

     */

    private void reconnect() {

        logger.info("[CanalClient reconnect] 重新连接 ...");

        runing = false;

        while (!runing) {

            try {

                connector = CanalConnectors.newSingleConnector(new InetSocketAddress(hostname, port), destination, username, password);

                connector.connect();

                connector.subscribe(".*\\..*");

                connector.rollback();

                runing = true;

            } catch (Exception e) {

                connector.disconnect();

                logger.error("[CanalClient] [reconnect] " + e.getMessage(), e);

                try {

                    Thread.sleep(3000);

                } catch (InterruptedException e1) {

                    logger.error(e1.getMessage(), e1);

                }

            }

        }

        logger.info("[CanalClient reconnect] 重新连接成功！");

        work();

    }

    private void process(List<Entry> entries) {

        try {

            for (Entry entry : entries) {

                if(logger.isDebugEnabled()){

                    logger.debug("mysql binlog : " + entry.getHeader().getLogfileName() + "=>" + entry.getHeader().getLogfileOffset());

                }

                /**

                 * 忽略 事务开启 、结束 ,query 的 binlog 内容

                 */

                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND || entry.getHeader().getEventType() == EventType.QUERY) {

                    continue;

                }

                logger.info("解析偏移量:" + entry.getHeader().getLogfileName() + "=>" + entry.getHeader().getLogfileOffset() + " ," +

                        "操作表[" + entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName() + "]," +

                        "变更类型[" + entry.getHeader().getEventType() + "]," +

                        "执行时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(entry.getHeader().getExecuteTime())));

                RowChange rowChange;

                try {

                    rowChange = RowChange.parseFrom(entry.getStoreValue());

                } catch (Exception e) {

                    logger.error("[CanalClient] [process] 解析RowChange事件错误: " + e.getMessage(), entry.toString());

                    continue;

                }

                log(entry.getHeader(), rowChange);

                if (gsonListeners.size() > 0) {

                    GsonEntry binlog = new GsonEntry(entry.getHeader(), rowChange);

                    for (CanalGsonListener listener : gsonListeners) {

                        listener.onBinlog(binlog);

                    }

                }

                if (binaryListeners.size() > 0) {

                    for (CanalBinaryListener listener : binaryListeners) {

                        listener.onBinlog(entry);

                    }

                }

            }

        } catch (Exception e) {

            logger.error("[CanalClient] [process] " + e.getMessage(), e);

        }

    }

    private void log(Header header, RowChange rowChange) {

        EventType eventType = rowChange.getEventType();

        if(logger.isDebugEnabled()){

            logger.debug(String.format("binlog[%s:%s], name[%s,%s], eventType : %s",

                    header.getLogfileName(), header.getLogfileOffset(),

                    header.getSchemaName(), header.getTableName(),

                    eventType));

        }

        for (RowData rowData : rowChange.getRowDatasList()) {

            if (eventType == EventType.DELETE) {

                log(rowData.getBeforeColumnsList());

            } else if (eventType == EventType.INSERT) {

                log(rowData.getAfterColumnsList());

            } else {

                log(rowData.getBeforeColumnsList());

                log(rowData.getAfterColumnsList());

            }

        }

    }

    private void log(List<Column> columns) {

        for (Column column : columns) {

            if(logger.isDebugEnabled()){

                logger.debug(new StringBuffer()

                        .append(column.getName()).append(" = ").append(column.getValue())

                        .append(" update[").append(column.getUpdated()).append("]").toString());

            }

        }

    }

}

kafka消息实体定义：

import java.sql.Timestamp

import com.alibaba.otter.canal.protocol.CanalEntry.EventType

/**

  * desc: BinlogEvent bean

  *

  * @author hz.lei  2018年03月07日 下午3:43

  */

case class BinlogEvent(schema: String, tableName: String, eventType: EventType, timestamp: Timestamp, before: String, after: String)

2. 客户端－订阅kafka消息获取想要的binlog变更

binlog监听类：

import com.today.binlog.BinlogEvent

import com.today.eventbus.annotation.{BinlogListener, KafkaConsumer}

import com.today.service.binlog.action._

import org.springframework.transaction.annotation.Transactional

import scala.collection.JavaConverters._

/**

  *

  * 描述: binlog 监听类

  *

  * @author hz.lei

  * @since 2018年03月08日 下午7:18

  */

@KafkaConsumer(groupId = "GOODS_0.0.1_EVENT", topic = "Binlog")

@Transactional(readOnly = true)

class GoodsBinlogListener {

  @BinlogListener

  def onBinlog(event: java.util.List[BinlogEvent]): Unit = {

    event.asScala.foreach(new GoodsOnBinlogAction(_).action())

  }

}

注意必须添加如下两个注解：

@KafkaConsumer(groupId = "GOODS_0.0.1_EVENT", topic = "Binlog")

@BinlogListener

添加注解扫描类：

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:soa="http://soa-springtag.dapeng.com/schema/service"

       xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"

       xsi:schemaLocation="http://www.springframework.org/schema/beans

        http://www.springframework.org/schema/beans/spring-beans.xsd

        http://soa-springtag.dapeng.com/schema/service

        http://soa-springtag.dapeng.com/schema/service/service.xsd

        http://www.springframework.org/schema/context

        http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">

    <bean id="postProcessor" class="com.today.eventbus.spring.MsgAnnotationBeanPostProcessor"/>

</beans>

注解扫描类定义:

import com.today.eventbus.ConsumerEndpoint;

import com.today.eventbus.annotation.BinlogListener;

import com.today.eventbus.annotation.KafkaConsumer;

import com.today.eventbus.annotation.KafkaListener;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.aop.framework.Advised;

import org.springframework.aop.support.AopUtils;

import org.springframework.beans.BeansException;

import org.springframework.beans.factory.BeanFactory;

import org.springframework.beans.factory.BeanFactoryAware;

import org.springframework.beans.factory.SmartInitializingSingleton;

import org.springframework.beans.factory.config.BeanPostProcessor;

import org.springframework.beans.factory.support.BeanDefinitionBuilder;

import org.springframework.beans.factory.support.DefaultListableBeanFactory;

import org.springframework.core.MethodIntrospector;

import org.springframework.core.Ordered;

import org.springframework.core.annotation.AnnotationUtils;

import org.springframework.util.ReflectionUtils;

import com.today.eventbus.utils.Constant;

import java.lang.reflect.Method;

import java.util.*;

/**

 * 描述: MsgAnnotationBeanPostProcessor bean 后处理器，扫描自定义注解 @KafkaListener

 *

 * @author hz.lei

 * @see KafkaListenerRegistrar,BeanFactory,BeanPostProcessor,SmartInitializingSingleton

 * @since 2018年03月01日 下午9:36

 */

public class MsgAnnotationBeanPostProcessor implements BeanPostProcessor, BeanFactoryAware, Ordered, SmartInitializingSingleton {

    /**

     * logger

     */

    private final Logger logger = LoggerFactory.getLogger(getClass());

    /**

     * hold beanFactory ,real impl is {@link DefaultListableBeanFactory}

     * for create bean dynamically, bean {@link KafkaListenerRegistrar}

     */

    private BeanFactory beanFactory;

    /**

     * 处理 kafka 消费者 的注册与创建

     */

    private KafkaListenerRegistrar registrar;

    /**

     * beanFactory 回调，让bean持有容器的引用

     *

     * @param beanFactory

     * @throws BeansException

     */

    @Override

    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {

        this.beanFactory = beanFactory;

        createKafkaRegistryBean();

    }

    /**

     * 动态创建bean KafkaListenerRegistrar

     */

    private void createKafkaRegistryBean() {

        // 获取bean工厂并转换为DefaultListableBeanFactory

        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) beanFactory;

        // 通过BeanDefinitionBuilder创建bean定义

        BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(KafkaListenerRegistrar.class);

        // 注册bean

        defaultListableBeanFactory.registerBeanDefinition(Constant.KAFKA_LISTENER_REGISTRAR_BEAN_NAME, beanDefinitionBuilder.getRawBeanDefinition());

        this.registrar = (KafkaListenerRegistrar) beanFactory.getBean(Constant.KAFKA_LISTENER_REGISTRAR_BEAN_NAME);

    }

    /**

     * 所有单例 bean 初始化完成后，调用此方法

     */

    @Override

    public void afterSingletonsInstantiated() {

        this.registrar.afterPropertiesSet();

    }

    /**

     * 实例化及依赖注入完成后、在任何初始化代码（比如配置文件中的init-method）调用之前调用

     *

     * @param bean

     * @param beanName

     * @return

     * @throws BeansException

     */

    @Override

    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {

        return bean;

    }

    /**

     * 实例化及依赖注入完成后、在任何初始化代码（比如配置文件中的init-method）调用之后调用

     *

     * @param bean

     * @param beanName

     * @return

     * @throws BeansException

     */

    @Override

    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

        logger.debug("access to postProcessAfterInitialization bean {}, beanName {}", bean, beanName);

        Class<?> targetClass = AopUtils.getTargetClass(bean);

        //获取类上是否有注解 @KafkaConsumer

        Optional<KafkaConsumer> kafkaConsumer = findListenerAnnotations(targetClass);

        //类上是否有注解

        final boolean hasKafkaConsumer = kafkaConsumer.isPresent();

        if (hasKafkaConsumer) {

            //方法列表 ，查找方法上标有 @KafkaListener 的注解

            Map<Method, Set<KafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,

                    (MethodIntrospector.MetadataLookup<Set<KafkaListener>>) method -> {

                        Set<KafkaListener> listenerMethods = findListenerAnnotations(method);

                        return (!listenerMethods.isEmpty() ? listenerMethods : null);

                    });

            //查找方法上标有 @BinlogListener 的注解

            Map<Method, Set<BinlogListener>> binlogMethods = MethodIntrospector.selectMethods(targetClass,

                    (MethodIntrospector.MetadataLookup<Set<BinlogListener>>) method -> {

                        Set<BinlogListener> listenerMethods = findBinlogListenerAnnotations(method);

                        return (!listenerMethods.isEmpty() ? listenerMethods : null);

                    });

            if (annotatedMethods.isEmpty() && binlogMethods.isEmpty()) {

                throw new IllegalArgumentException("@KafkaConsumer found on class type , " +

                        "but no @KafkaListener or @BinlogListener found on the method ,please set it on the method");

            }

            if (!annotatedMethods.isEmpty() && !binlogMethods.isEmpty()) {

                throw new IllegalArgumentException("@KafkaListener or @BinlogListener only one could on the same  bean class");

            }

            if (!annotatedMethods.isEmpty()) {

                // Non-empty set of methods

                for (Map.Entry<Method, Set<KafkaListener>> entry : annotatedMethods.entrySet()) {

                    Method method = entry.getKey();

                    for (KafkaListener listener : entry.getValue()) {

                        // process annotation information

                        processKafkaListener(kafkaConsumer.get(), listener, method, bean, beanName);

                    }

                }

                logger.info("there are {} methods have @KafkaListener on This bean ", binlogMethods.size());

            }

            if (!binlogMethods.isEmpty()) {

                // Non-empty set of methods

                for (Map.Entry<Method, Set<BinlogListener>> entry : binlogMethods.entrySet()) {

                    Method method = entry.getKey();

                    // process annotation information

                    processBinlogListener(kafkaConsumer.get(), method, bean);

                }

                logger.info("there are {} methods have @BinlogListener on This bean ", binlogMethods.size());

            }

            if (this.logger.isDebugEnabled()) {

                this.logger.debug(annotatedMethods.size() + " @KafkaListener methods processed on bean '"

                        + beanName + "': " + annotatedMethods);

            }

        } else {

            this.logger.info("No @KafkaConsumer annotations found on bean type: " + bean.getClass());

        }

        return bean;

    }

    /**

     * 扫描 bean 类上 是否有注解 @KafkaConsumer,只有有此注解才说明 是kafka message 消费者

     */

    private Optional<KafkaConsumer> findListenerAnnotations(Class<?> clazz) {

        KafkaConsumer ann = AnnotationUtils.findAnnotation(clazz, KafkaConsumer.class);

        return Optional.ofNullable(ann);

    }

    /**

     * 扫描bean 方法上 是否有注解 @KafkaListener

     *

     * @param method

     * @return

     */

    private Set<KafkaListener> findListenerAnnotations(Method method) {

        Set<KafkaListener> listeners = new HashSet<>();

        KafkaListener ann = AnnotationUtils.findAnnotation(method, KafkaListener.class);

        if (ann != null) {

            listeners.add(ann);

        }

        return listeners;

    }

    /**

     * 扫描bean 方法上 是否有注解 @BinlogListener

     *

     * @param method

     * @return

     */

    private Set<BinlogListener> findBinlogListenerAnnotations(Method method) {

        Set<BinlogListener> listeners = new HashSet<>();

        BinlogListener ann = AnnotationUtils.findAnnotation(method, BinlogListener.class);

        if (ann != null) {

            listeners.add(ann);

        }

        return listeners;

    }

    /**

     * 处理有 @KafkaListener 注解的 方法上注解元信息，封装成 consumerEndpoint，注册

     *

     * @param consumer

     * @param listener

     * @param method

     * @param bean

     * @param beanName

     */

    protected void processKafkaListener(KafkaConsumer consumer, KafkaListener listener, Method method, Object bean, String beanName) {

        Method methodToUse = checkProxy(method, bean);

        ConsumerEndpoint endpoint = new ConsumerEndpoint();

        endpoint.setMethod(methodToUse);

        endpoint.setBean(bean);

        endpoint.setParameterTypes(Arrays.asList(method.getParameterTypes()));

        // class annotation information

        endpoint.setGroupId(consumer.groupId());

        endpoint.setTopic(consumer.topic());

        endpoint.setKafkaHostKey(consumer.kafkaHostKey());

        // method annotation information

        endpoint.setSerializer(listener.serializer());

        //session timeout

        if (consumer.sessionTimeout() < Constant.DEFAULT_SESSION_TIMEOUT) {

            throw new RuntimeException("抛出该异常原因为: kafkaConsumer session 超时时间设置太小 ,请设置至少为 10000L 以上,单位为 ms(毫秒)");

        }

        endpoint.setTimeout(consumer.sessionTimeout());

        this.registrar.registerEndpoint(endpoint);

    }

    private void processBinlogListener(KafkaConsumer consumer, Method method, Object bean) {

        Method methodToUse = checkProxy(method, bean);

        ConsumerEndpoint endpoint = new ConsumerEndpoint();

        endpoint.setMethod(methodToUse);

        endpoint.setBean(bean);

        // class annotation information

        endpoint.setGroupId(consumer.groupId());

        endpoint.setTopic(consumer.topic());

        endpoint.setKafkaHostKey(consumer.kafkaHostKey());

        // method annotation information

        endpoint.setBinlog(true);

        this.registrar.registerEndpoint(endpoint);

    }

    /**

     * 获取目标方法，如果是代理的，获得其目标方法

     *

     * @param methodArg

     * @param bean

     * @return

     */

    private Method checkProxy(Method methodArg, Object bean) {

        Method method = methodArg;

        if (AopUtils.isJdkDynamicProxy(bean)) {

            try {

                // Found a @KafkaListener method on the target class for this JDK proxy ->

                // is it also present on the proxy itself?

                method = bean.getClass().getMethod(method.getName(), method.getParameterTypes());

                Class<?>[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces();

                for (Class<?> iface : proxiedInterfaces) {

                    try {

                        method = iface.getMethod(method.getName(), method.getParameterTypes());

                        break;

                    } catch (NoSuchMethodException noMethod) {

                    }

                }

            } catch (SecurityException ex) {

                ReflectionUtils.handleReflectionException(ex);

            } catch (NoSuchMethodException ex) {

                throw new IllegalStateException(String.format(

                        "@KafkaListener method '%s' found on bean target class '%s', " +

                                "but not found in any interface(s) for bean JDK proxy. Either " +

                                "pull the method up to an interface or switch to subclass (CGLIB) " +

                                "proxies by setting proxy-target-class/proxyTargetClass " +

                                "attribute to 'true'", method.getName(), method.getDeclaringClass().getSimpleName()), ex);

            }

        }

        return method;

    }

    @Override

    public int getOrder() {

        return LOWEST_PRECEDENCE;

    }

}

处理 binlog 缓存监听事件:

import com.github.dapeng.core.SoaException;

import com.today.eventbus.common.MsgConsumer;

import com.today.eventbus.common.retry.BinlogRetryStrategy;

import com.today.eventbus.ConsumerEndpoint;

import com.today.eventbus.config.KafkaConfigBuilder;

import com.today.eventbus.serializer.KafkaIntDeserializer;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.common.serialization.ByteArrayDeserializer;

import java.lang.reflect.InvocationTargetException;

import java.util.List;

import java.util.Properties;

import java.util.stream.Collectors;

/**

 * 描述: 处理 binlog 缓存 监听 事件

 *

 * @author hz.lei

 * @since 2018年03月07日 上午1:42

 */

public class BinlogKafkaConsumer extends MsgConsumer<Integer, byte[], ConsumerEndpoint> {

    /**

     * @param kafkaHost host1:port1,host2:port2,...

     * @param groupId

     * @param topic

     */

    public BinlogKafkaConsumer(String kafkaHost, String groupId, String topic) {

        super(kafkaHost, groupId, topic);

    }

    @Override

    protected void init() {

        logger.info("[KafkaConsumer] [init] " +

                "kafkaConnect(" + kafkaConnect +

                ") groupId(" + groupId +

                ") topic(" + topic + ")");

        KafkaConfigBuilder.ConsumerConfiguration builder = KafkaConfigBuilder.defaultConsumer();

        final Properties props = builder.bootstrapServers(kafkaConnect)

                .group(groupId)

                .withKeyDeserializer(KafkaIntDeserializer.class)

                .withValueDeserializer(ByteArrayDeserializer.class)

                .withOffsetCommitted(false)

                .excludeInternalTopic(false)

                .maxPollSize("100")

                .build();

        consumer = new KafkaConsumer<>(props);

    }

    @Override

    protected void buildRetryStrategy() {

        retryStrategy = new BinlogRetryStrategy();

    }

    @Override

    protected void dealMessage(ConsumerEndpoint consumer, byte[] value, Integer keyId) throws SoaException {

        List<BinlogEvent> binlogEvents = BinlogMsgProcessor.process(value);

        // > 0 才处理

        if (binlogEvents.size() > 0) {

            try {

                consumer.getMethod().invoke(consumer.getBean(), binlogEvents);

            } catch (IllegalAccessException e) {

                logger.error("BinlogConsumer::实例化@BinlogListener 注解的方法 出错", e);

            } catch (InvocationTargetException e) {

                throwRealException(e, consumer.getMethod().getName());

            }

            logger.info("BinlogConsumer::[dealMessage(id: {})] end, method: {}, groupId: {}, topic: {}, bean: {}",

                    keyId, consumer.getMethod().getName(), groupId, topic, consumer.getBean());

        }

    }

}

匹配binlog变动，获取变更前后信息：

import com.alibaba.otter.canal.protocol.CanalEntry

import com.today.binlog.BinlogEvent

import com.today.service.GoodsDataSource

import com.today.service.binlog.bean.SkuBean

import com.today.service.commons.cache.MemcacheProcessor

import com.today.service.commons.cache.dto.RedisBean

import redis.clients.jedis.JedisPool

import spray.json._

import wangzx.scala_commons.sql._

import com.today.service.commons.help.BizHelp.withJedis

import com.today.service.commons.help.SqlHelp

import com.today.service.commons.util.DateTools

import com.today.service.commons.`implicit`.Implicits._

import scala.collection.JavaConverters._

class GoodsOnBinlogAction(binlogEvent: BinlogEvent) extends MemcacheProcessor("goods_db_sku") {

  def action() = {

    logger.info(s"${getClass.getSimpleName} onBinlog ")

    logger.info(s"binlogEvent:$binlogEvent")

    binlogEvent match {

      case BinlogEvent("goods_db", "sku", CanalEntry.EventType.INSERT, timestamp, before,

      json"""{"id": $id,"sku_no":${skuNoJsValue}}"""

      ) => {

        logger.info(s"${getClass.getSimpleName} onInsert...")

        val skuNo = toStringValue(skuNoJsValue)

        val skuBean = reloadBySkuNo(skuNo)

        if (skuBean.isDefined) insertSortSet(skuBean.get)

      }

      case BinlogEvent("goods_db", "sku", CanalEntry.EventType.UPDATE, timestamp, json"""{"id": $beforeSkuId,"sku_no":${beforeSkuNoJsValue}}""",

      json"""{"id": $afterSkuId,"sku_no":${afterSkuNoJsValue}}"""

      ) => {

        logger.info(s"${getClass.getSimpleName} onUpdate...")

        val beforeSkuNo = toStringValue(beforeSkuNoJsValue)

        val afterSkuNo = toStringValue(afterSkuNoJsValue)

        if (!beforeSkuNo.equals(afterSkuNo)) {

          deleteSortSet(beforeSkuNo)

        }

        val skuBean = reloadBySkuNo(afterSkuNo)

        if (skuBean.isDefined) updateSortSet(skuBean.get.primaryKey, skuBean.get)

      }

      case BinlogEvent("goods_db", "sku", CanalEntry.EventType.DELETE, timestamp, json"""{"id": $id,"sku_no":${skuNo}}""", after) => {

        logger.info(s"${getClass.getSimpleName} onDelete...")

        deleteSortSet(toStringValue(skuNo))

      }case _ =>

    }

  }

}