准备
找到我们所需的安装包文件
关于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
创建目录data目录
用于存放持久化数据的地方
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
此忽略
配置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消息队列
安装kibana和es
如何安装,可以参考我的这篇文章,有介绍如何安装es和kibana以及ik分词器
https://www.jianshu.com/p/f52d9c843bd8
启动后的如下
建立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 -->
<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-elasticsearch</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.70</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.12</version>
<scope>provided</scope>
</dependency>
</dependencies>
配置文件
# kafka
spring:
data:
elasticsearch:
####集群名称
cluster-name: docker-cluster
####地址
cluster-nodes: 192.168.137.5:9300
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
实体类代码
@Document(indexName = "student", type = "student")
@Data
public class Student {
private Integer id;
private String name;
}
mapper类
public interface StudentMapper extends ElasticsearchRepository<Student, Long> {
}
kafka消费者代码
@Component
public class MembetKafkaConsumer {
@Autowired
private StudentMapper studentMapper;
@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);
Student student = dataObject.toJavaObject(Student.class);
//我这里为了方便,检测类型为insert或者update,都统一为save,实际到es则直接覆盖掉了
switch (type) {
case "UPDATE":
case "INSERT":
studentMapper.save(student);
break;
case "DELETE":
studentMapper.delete(student);
break;
}
}
}
// @KafkaListener(topics = "sunny-topic")
// public void onMessage(String message){
// //insertIntoDb(buffer);//这里为插入数据库代码
// System.out.println(message);
// }
}
然后在数据库中,修改name字段
kafka立马接收到消息
kibana检索同步到es的数据
到此基于canal+kafka实现了mysql与es数据同步啦
网友评论