maxwell收集binlog，不同网络环境通过nginx发送数据到kafka集群
咻咻ing 2020-04-10
但有时，应用部署在异地环境，mysql数据库的变化通过maxwell无法直接发送到数据中心进行解析和数据同步，本次使用ngix方式作为代理服务器，收集maxwell发送的json数据后，发送到后端的kafka集群。
架构如下：
1，多个应用平台分布在不同地域内，远端mysql数据库，可以访问互联网。
2，在本地数据中心，使用nginx服务代理多台kafka集群。
3，把nginx服务器ip通过公网IP+端口映射，可以通过公网ip进行访问nginx。
在通过以上架构设计后，但是maxwell是不支持发送到http服务，只支持kafka、redis等。
在查阅maxwell官网后，发现有自定义producer方式，本次即使用自定义方式解决maxwell通过post发送json到nginx。
（文中代码中颜色是系统自带，不需要过度注意。）
一、代码开发工作
1，使用idea，建立maven工程，添加pom依赖，主要设计http相关
<dependency>
<groupId>commons-httpclient</groupId>
<artifactId>commons-httpclient</artifactId>
<version>3.1</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.2</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpasyncclient</artifactId>
<version>4.1.2</version>
</dependency>2，手动添加maxwell-1.22.3.jar文件到项目中。
3，创建HttpUtil类，用于调用发送post请求
package com.test.utils;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
public class HttpUtil {
public void doPost(String url, String json){
CloseableHttpClient httpclient = HttpClientBuilder.create().build();
HttpPost post = new HttpPost(url);
try {
StringEntity s = new StringEntity(json.toString());
s.setContentEncoding("UTF-8");
s.setContentType("application/json");//发送json数据需要设置contentType
post.setEntity(s);
HttpResponse res = httpclient.execute(post);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}4，创建CustomProducer自定义类，继承AbstractProducer
package com.test.producerfactory;
import com.test.utils.HttpUtil;
import com.zendesk.maxwell.MaxwellContext;
import com.zendesk.maxwell.producer.AbstractProducer;
import com.zendesk.maxwell.producer.EncryptionMode;
import com.zendesk.maxwell.producer.MaxwellOutputConfig;
import com.zendesk.maxwell.row.RowMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
public class CustomProducer extends AbstractProducer {
private final String headerFormat;
private final Collection<RowMap> txRows = new ArrayList<>();
private final HttpUtil httpUtil=new HttpUtil();
private static MaxwellOutputConfig config=new MaxwellOutputConfig();
private String url="";
private String server_id="0";
private String encrypt=null;
private String secretKey=null;
public CustomProducer(MaxwellContext context) {
super(context);
// this property would be ‘custom_producer.header_format‘ in config.properties
headerFormat = context.getConfig().customProducerProperties.getProperty("header_format", "Transaction: %xid% >>>\n");
//从maxwell的配置文件中获取配置信息
server_id=context.getConfig().customProducerProperties.getProperty("server_id");
url=context.getConfig().customProducerProperties.getProperty("url");
encrypt=context.getConfig().customProducerProperties.getProperty("encrypt");
secretKey=context.getConfig().customProducerProperties.getProperty("secretKey");
// 配置输出json字段包含serverID
config.includesServerId=true;
//配置是否加密数据
if (encrypt.equals("data")){
config.encryptionMode= EncryptionMode.ENCRYPT_DATA;
config.secretKey=secretKey;
}else if (encrypt.equals("all")){
config.encryptionMode= EncryptionMode.ENCRYPT_ALL;
config.secretKey=secretKey;
}
}
@Override
public void push(RowMap r) throws Exception
{
// filtering out DDL and heartbeat rows
if(!r.shouldOutput(outputConfig)) {
// though not strictly necessary (as skipping has no side effects), we store our position,
// so maxwell won‘t have to "re-skip" this position if crashing and restarting.
context.setPosition(r.getPosition());
return;
}
//设置serverID
r.setServerId(Long.parseLong(server_id));
// store uncommitted row in buffer
txRows.add(r);
if(r.isTXCommit()) {
// This row is the final and closing row of a transaction. Stream all rows of buffered
// transaction to stdout
// System.out.print(headerFormat.replace("%xid%", r.getXid().toString()));
txRows.stream().map(CustomProducer::toJSON).forEach(string -> httpUtil.doPost(url,string));
txRows.clear();
// rows ++;
// Only now, after finally having "persisted" all buffered rows to stdout is it safe to
// store the producers position.
context.setPosition(r.getPosition());
//
}
}
private static String toJSON(RowMap row) {
try {
return row.toJSON(config);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}5，创建CustomProducerFactory类
package com.test.producerfactory;
import com.zendesk.maxwell.MaxwellContext;
import com.zendesk.maxwell.producer.AbstractProducer;
import com.zendesk.maxwell.producer.ProducerFactory;
public class CustomProducerFactory implements ProducerFactory{
@Override
public AbstractProducer createProducer(MaxwellContext context) {
return new CustomProducer(context);
}
}6，使用idea工具打包data_sync.jar文件传到远端的maxwell的lib目录下。
二、配置工作
配置工作主要分nginx和maxwell配置，下面分别介绍配置项。
1，nginx配置
Nginx在下载后，进行源码编译，需要添加kafka支持的插件
[ nginx]#
./configure --add-module=/usr/local/src/ngx_kafka_module --add-module=/usr/logcal/nginx_tcp_proxy_module
nginx的安装方式不做介绍，在安装完nginx后，在/usr/local/nginx/conf目录下，编辑nginx.conf文件
#user nobody;
worker_processes 1;
error_log logs/error.log;
error_log logs/error.log notice;
error_log logs/error.log info;
pid logs/nginx.pid;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
kafka;
kafka_broker_list host2:9092 host3:9092 host4:9092;
server {
listen 19090;
server_name localhost;
location / {
root html;
kafka_topic test1;
index index.html index.htm;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}
}其中kafka_topic是接收数据后发送到指定的topic。
kafka_broker_list：即是kafka的broker节点和端口，此处因为配置了host解析，所以用了主机名。
在nginx配置完成后，reload配置，可以使用与kafka、nginx不同网段的服务器，使用以下命令测试nginx是否配通：
[ ~]# curl http://58.30.1.xxx:19007/ -d "aaaaaa"
在内网kafka集群中，使用以下命令查看kafka是否能接受到数据：
[ ~]#kafka-console-consumer --bootstrap-server kafkahost:9092 --topic test1
当kafka集群中接收到数据后，表示http发送的数据经过nginx 转发到kafka集群。
2，maxwell配置，可以通过官网下载maxwell软件，解压到/opt/maxwell下
（具体maxwell的安装和启动方式我在上一篇中已经有详细介绍）
使用自定义生产消费者，在解压后的maxwell上传依赖的data_sync.jar报到/opt/maxwell/lib目录下。
在/opt/maxwell目录下创建一个config.properties文件，写入指定配置：
vim config.properties
#[mysql] user=maxwell password=123456 host=hadoop1 port=3306 #[producer] output_server_id=true custom_producer.factory=com.test.producerfactory.CustomProducerFactory custom_producer.server_id=23 custom_producer.url=http://58.30.1.XX:19007/ custom_producer.encrypt=data custom_producer.secretKey=0f1b122303xx44123
配置项说明：
user： #连接mysql用户名
password：#连接mysql的密码
host： # mysql的主机名（IP地址）
port：#mysql端口
output_server_id： #输出server_id，用于标识哪个区域平台的数据
custom_producer.factory： #自定义生产消费类
custom_producer.server_id： #定义的server_id，与my.cnf中的server_id一致
custom_producer.url： #数据中心对外开放的url
custom_producer.encrypt： #加密方式，data、all、none
custom_producer.secretKey： #秘钥值，通过数据中心点分配的秘钥值，与server_id一一对应
如果配置了数据加密，在接收到数据后，还需要进一步的解密后，才能获取binlog数据，解密的方式后续会写方法。
以上配置完成后，即可以启动maxwell，开启同步数据到本地数据中心，当数据同步到本地kafka集群后，可以使用flink、spark streaming接收做进一步处理。
