準(zhǔn)備

找到我們所需的安裝包文件
關(guān)于Apache的相關(guān)包,都可以在這個(gè)網(wǎng)站找到啦

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 文件, 在文件末尾添加以下環(huán)境變量配置

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

運(yùn)行以下命令使環(huán)境變量生效

source /etc/profile

重命名配置文件

初次使用需要將config下zoo_sample.cfg 重命名為 zoo.cfg

cd zookeeper/conf/

mv zoo_sample.cfg zoo.cfg

創(chuàng)建目錄data目錄

用于存放持久化數(shù)據(jù)的地方

mkdir data

修改配置文件

修改zoo.cfg中的datadir路徑為將剛剛創(chuàng)建的data目錄的地址

啟動(dòng)zookeeper服務(wù)

/usr/local/software/zookeeper/bin

./zkServer.sh start

當(dāng)然也可以通過status命令,來查看zk是否成功運(yùn)運(yùn)行,以及什么模式進(jìn)行運(yùn)行等

./zkServer.sh status

好,如下圖就代表我們成功連接啟動(dòng)了zk服務(wù)器啦~

安裝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

創(chuàng)建logs目錄

cd kafka

mkdir logs

修改配置文件

vim /usr/local/software/kafka/config/server.properties 修改參數(shù)

啟動(dòng)kafka Server

啟動(dòng)kafka

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

關(guān)閉kafka

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

查看kafka是否關(guān)閉

jps

查看所有topic

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

查看指定topic下的數(shù)據(jù)

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

如何判斷是否啟動(dòng)成功,請(qǐng)看下圖

安裝mysql

安裝mysql

此忽略

配置mysql

vim /etc/my.cnf

log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定義甲雅，不要和 canal 的 slaveId 重復(fù)

授權(quán) canal 鏈接 MySQL 賬號(hào)具有作為 MySQL slave 的權(quán)限, 如果已有賬戶可直接 grant

drop user 'canal'@'%';    ##注意,這里如果沒有創(chuàng)建canal賬號(hào),則會(huì)報(bào)錯(cuò),所以沒有創(chuàng)建則直接忽略這一步
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則開啟

檢查權(quán)限

一定要檢查mysql user 權(quán)限為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

解壓完成后,可以看到如下結(jié)構(gòu)

配置修改

vi conf/example/instance.properties

修改canal 配置文件

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

注釋寫著暫時(shí)支持三種監(jiān)聽模式,默認(rèn)是tcp模式,我們選擇KafKa監(jiān)聽

更改為kafka的連接地址,或集群地址

改完后,進(jìn)入bin目錄重啟canalServer端
重啟完了后,如何查看是否集成kafka成功了呢,很簡(jiǎn)單
直接進(jìn)入zk里面查看kafka的主題,是否有我們剛才在server端定義的名稱即可確定是否集成成功
注意的是:如果重啟后沒有發(fā)現(xiàn)zk里面有自己的topic主題赴背，可能是懶加載的原因,可以通過修改數(shù)據(jù)庫(kù)來實(shí)現(xiàn)同步數(shù)據(jù),這個(gè)時(shí)候zk就會(huì)有自己的Topic了

啟動(dòng)canal

bin/startup.sh

查看是否啟動(dòng)成功

通過查看日志
在canal目錄下的/logs/example/example.log日志

那么,zk伦忠、kafka曙咽、canal都啟動(dòng)了
我們新建一個(gè)數(shù)據(jù)庫(kù),并新增一個(gè)表,添加一條數(shù)據(jù),看數(shù)據(jù)是否會(huì)被監(jiān)聽到

查看某個(gè)topic的所有消息

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

消息已成功達(dá)到kafka消息隊(duì)列

建立SpringBoot項(xiàng)目

項(xiàng)目結(jié)構(gòu)

添加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服務(wù)器地址(可以多個(gè))
    bootstrap-servers: 192.168.137.5:9092
    consumer:
      # 指定一個(gè)默認(rèn)的組名
      group-id: kafka2
      # earliest:當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無(wú)提交的offset時(shí)，從頭開始消費(fèi)
      # latest:當(dāng)各分區(qū)下有已提交的offset時(shí)境蔼，從提交的offset開始消費(fèi)灶平；無(wú)提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)
      # none:topic各分區(qū)都存在已提交的offset時(shí)箍土，從offset后開始消費(fèi)逢享；只要有一個(gè)分區(qū)不存在已提交的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消費(fèi)者代碼

@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() + "," +
                "分區(qū)位置:" + consumer.partition()
                + ", 下標(biāo)" + 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);//這里為插入數(shù)據(jù)庫(kù)代碼
//        System.out.println(message);
//    }

}

redis工具類

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

    /**
     * 存放string類型
     *
     * @param key     key
     * @param data    數(shù)據(jù)
     * @param timeout 超時(shí)間
     */
    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 數(shù)據(jù)
     */
    public void setString(String key, String data) {
        setString(key, data, null);
    }

    /**
     * 根據(jù)key查詢string類型
     *
     * @param key
     * @return
     */
    public String getString(String key) {
        String value = stringRedisTemplate.opsForValue().get(key);
        return value;
    }

    /**
     * 根據(jù)對(duì)應(yīng)的key刪除key
     *
     * @param key
     */
    public boolean delKey(String key) {
        return stringRedisTemplate.delete(key);
    }
}

數(shù)據(jù)庫(kù)數(shù)據(jù)發(fā)生改變,kafka立馬消費(fèi)消息

Redis同步數(shù)據(jù)

本文就到這,如有疑問,評(píng)論一起討論