准备

找到我们所需的安装包文件
关于Apache的相关包,都可以在这个网站找到啦

http://mirrors.hust.edu.cn/apache/

安装JDK

此处略过

安装zookeeper

下载源码包,并解压

wget http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

tar -zxvf zookeeper-3.4.14.tar.gz

mv zookeeper-3.4.14/ zookeeper

修改配置文件

编辑 /etc/profile 文件, 在文件末尾添加以下环境变量配置

export ZOOKEEPER_HOME=/usr/local/zookeeper
export PATH=$PATH:$ZOOKEEPER_HOME/bin

运行以下命令使环境变量生效

source /etc/profile

重命名配置文件

初次使用需要将config下zoo_sample.cfg 重命名为 zoo.cfg

cd zookeeper/conf/

mv zoo_sample.cfg zoo.cfg

用于存放持久化数据的地方

mkdir data

修改配置文件

修改zoo.cfg中的datadir路径为将刚刚创建的data目录的地址

启动zookeeper服务

/usr/local/software/zookeeper/bin

./zkServer.sh start

当然也可以通过status命令,来查看zk是否成功运运行,以及什么模式进行运行等

./zkServer.sh status

好,如下图就代表我们成功连接启动了zk服务器啦~

安装kafka

下载源码包,并解压

wget http://mirrors.hust.edu.cn/apache/kafka/2.2.2/kafka_2.11-2.2.2.tgz

tar tar -zxvf kafka_2.11-2.2.2.tgz

mv kafka_2.11-2.2.2/ kafka

创建logs目录

cd kafka

mkdir logs

修改配置文件

vim /usr/local/software/kafka/config/server.properties 修改参数

启动kafka Server

启动kafka

bin/kafka-server-start.sh  -daemon  config/server.properties &

关闭kafka

bin/kafka-server-stop.sh  -daemon  config/server.properties &

查看kafka是否关闭

jps

查看所有topic

bin/kafka-topics.sh --list --zookeeper 192.168.137.5:2181

查看指定topic下的数据

bin/kafka-console-consumer.sh --bootstrap-server 192.168.137.5:9092  --from-beginning --topic sunny-topic

如何判断是否启动成功,请看下图

安装mysql

此忽略

配置mysql

vim /etc/my.cnf

log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

drop user 'canal'@'%';    ##注意,这里如果没有创建canal账号,则会报错,所以没有创建则直接忽略这一步
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
grant all privileges on *.* to 'canal'@'%' identified by 'canal';
flush privileges;

重启mysql
service mysqld restart

查看是否成功开启binlog日志

show variables like '%log_bin%';

log_bin为ON则开启

检查权限

一定要检查mysql user 权限为y

安装Canal

下载源码包,并解压

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-1/canal.deployer-1.1.5-SNAPSHOT.tar.gz

mkdir canal

tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz -C canal

解压完成后,可以看到如下结构

配置修改

vi conf/example/instance.properties

修改canal 配置文件

vim /usr/local/software/canal/conf/canal.properties

注释写着暂时支持三种监听模式,默认是tcp模式,我们选择KafKa监听

更改为kafka的连接地址,或集群地址

改完后,进入bin目录重启canalServer端
重启完了后,如何查看是否集成kafka成功了呢,很简单
直接进入zk里面查看kafka的主题,是否有我们刚才在server端定义的名称即可确定是否集成成功
注意的是:如果重启后没有发现zk里面有自己的topic主题，可能是懒加载的原因,可以通过修改数据库来实现同步数据,这个时候zk就会有自己的Topic了

启动canal

bin/startup.sh

查看是否启动成功

通过查看日志
在canal目录下的/logs/example/example.log日志

那么,zk、kafka、canal都启动了
我们新建一个数据库,并新增一个表,添加一条数据,看数据是否会被监听到

查看某个topic的所有消息

bin/kafka-console-consumer.sh --bootstrap-server 192.168.137.5:9092  --from-beginning --topic sunny-topic

消息已成功达到kafka消息队列

建立SpringBoot项目

项目结构

添加maven依赖

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.11.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>


    <dependencies>
        <!-- springBoot集成kafka -->
        <!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
<!--            <version>2.5.1.RELEASE</version>-->
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.70</version>
        </dependency>

    </dependencies>

配置文件

# kafka
spring:
  kafka:
    # kafka服务器地址(可以多个)
    bootstrap-servers: 192.168.137.5:9092
    consumer:
      # 指定一个默认的组名
      group-id: kafka2
      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 缓存容量
      buffer-memory: 524288
  redis:
    host: 127.0.0.1
    #    password:
    port: 6379
    database: 0

kafka消费者代码

@Component
public class MembetKafkaConsumer {
    @Autowired
    private RedisUtil redisUtil;

    @KafkaListener(topics = "sunny-topic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        System.out.println("topic名称:" + consumer.topic() + ",key:" +
                consumer.key() + "," +
                "分区位置:" + consumer.partition()
                + ", 下标" + consumer.offset() + "," + consumer.value());
        String json = (String) consumer.value();
        JSONObject jsonObject = JSONObject.parseObject(json);
        String type = jsonObject.getString("type");
        String pkNames = jsonObject.getJSONArray("pkNames").getString(0);
        JSONArray data = jsonObject.getJSONArray("data");
        for (int i = 0; i < data.size(); i++) {
            JSONObject dataObject = data.getJSONObject(i);
            String key = dataObject.getString(pkNames);
            switch (type) {
                case "UPDATE":
                case "INSERT":
                    redisUtil.setString(key, dataObject.toJSONString());
                    break;
                case "DELETE":
                    redisUtil.delKey(key);
                    break;
            }
        }

    }

//    @KafkaListener(topics = "sunny-topic")
//    public void onMessage(String message){
//        //insertIntoDb(buffer);//这里为插入数据库代码
//        System.out.println(message);
//    }

}

redis工具类

@Component
public class RedisUtil {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 存放string类型
     *
     * @param key     key
     * @param data    数据
     * @param timeout 超时间
     */
    public void setString(String key, String data, Long timeout) {
        stringRedisTemplate.opsForValue().set(key, data);
        if (timeout != null) {
            stringRedisTemplate.expire(key, timeout, TimeUnit.SECONDS);
        }
    }

    /**
     * 存放string类型
     *
     * @param key  key
     * @param data 数据
     */
    public void setString(String key, String data) {
        setString(key, data, null);
    }

    /**
     * 根据key查询string类型
     *
     * @param key
     * @return
     */
    public String getString(String key) {
        String value = stringRedisTemplate.opsForValue().get(key);
        return value;
    }

    /**
     * 根据对应的key删除key
     *
     * @param key
     */
    public boolean delKey(String key) {
        return stringRedisTemplate.delete(key);
    }
}

数据库数据发生改变,kafka立马消费消息

Redis同步数据

本文就到这,如有疑问,评论一起讨论

基于阿里巴巴Canal框架,kafka监听协议实现Mysql与Redis数据同步

基于阿里巴巴Canal框架,kafka监听协议实现Mysql与Redis数据同步

准备

安装JDK

安装zookeeper

下载源码包,并解压

修改配置文件

重命名配置文件

创建目录data目录

修改配置文件

启动zookeeper服务

安装kafka

下载源码包,并解压

创建logs目录

修改配置文件

启动kafka Server

启动kafka

关闭kafka

查看kafka是否关闭

查看所有topic

查看指定topic下的数据

安装mysql

安装mysql

配置mysql

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

检查权限

安装Canal

下载源码包,并解压

配置修改

修改canal 配置文件

建立SpringBoot项目

项目结构

添加maven依赖

配置文件

kafka消费者代码

redis工具类