部署dmhs，实现同步复制

1、环境检查（源端）

需要检查源端的数据库是否开启了归档日志和逻辑日志，如果有开启，忽略下面的操作
Bash
1、检查归档日志和逻辑日志是否开启
su - dmdba
cd /dm8/dminstall/bin
./disql SYSDBA/SYSDBA@192.168.159.133:5236
检查归档配置的正确性，如果能查询到结果就是开启了
SELECT ARCH_DEST, ARCH_FILE_SIZE FROM SYS.V$DM_ARCH_INI WHERE ARCH_TYPE=‘LOCAL’ AND ARCH_IS_VALID=‘Y’;

#检查逻辑日志配置的正确性，查询出来的结果是1或者2是开启了。
SELECT PARA_VALUE FROM SYS.V$DM_INI WHERE PARA_NAME = ‘RLOG_APPEND_LOGIC’;
exit

2、如果上面的查询结果显示没有开启，执行下面的操作开启
vim /dm8/dmdata/SOURCE/dm.ini
ARCH_INI和RLOG_APPEND_LOGIC设置成1，表示开启
ARCH_INI = 1
RLOG_APPEND_LOGIC = 1

3、添加归档配置文件
vim /dm8/dmdata/SOURCE/dmarch.ini

内容如下：

[ARCHIVE_LOCAL1]
ARCH_TYPE = LOCAL
ARCH_DEST = /dm8/dmarch #归档目录
ARCH_FILE_SIZE = 128 #归档文件大小，单位 MB
ARCH_SPACE_LIMIT = 0 #空间大小限制，0 表示不限制

4、重启数据库
./DmServiceSOURCE stop
./DmServiceSOURCE start

5、执行上面的1步骤，检查是否开启成功

2、安装dmhs服务(源端和目的端都操作)（dmdba用户）

1、关闭防火墙
Bash
systemctl stop firewalld
systemctl disable firewalld
2、上传dmhs的bin、key文件
把下载好的bin、key文件上传到服务器的/opt目录下，并给bin文件授予执行权限。
Bash
chmod +x /opt/dmhs_V4.3.20_dm8_rev140201_rh6_64_20230916.bin

3、开始安装dmhs
Bash
su - dmdba
/opt/dmhs_V4.3.20_dm8_rev140201_rh6_64_20230916.bin -i
大部分配置使用默认的就行，直接回车

3、配置dmhs.hs

配置源端：
bash
su - dmdba
cd /dm8/dmhs
#复制一份bin目录，单独配置，当同一个服务上需要部署多个dmhs，可以方便区分
cp -r bin cptkafka
cd cptkafka
#如果是测试dmhs，在使用的过程中可能会缺少libdmoci.so动态库文件报错，执行下面的命令复制一份到数据库的bin目录下，如果是生产中，需要去达梦官方申请
cp -r stat/libdmoci.so /dm8/dminstall/bin

vim dmhs.hs

内容如下：

<?xml version=“1.0” encoding=“GB2312” standalone=“no”?>
<dmhs>
<base>
<lang>en</lang>
<mgr_port>5345</mgr_port>
<ckpt_interval>60</ckpt_interval>
<siteid>1</siteid>
<version>2.0</version>
</base>
<cpt>
<db_type>DM8</db_type>
<db_server>192.168.159.133</db_server>
<db_user>APP</db_user>
<db_pwd>Dameng123</db_pwd>
<db_port>5236</db_port>
<idle_time>10</idle_time>
<ddl_mask>0</ddl_mask>
<cpt_mask>PARSE:POST:REG_OP2</cpt_mask>
<n2c>0</n2c>
<update_fill_flag>3</update_fill_flag>
<set_heartbeat>1</set_heartbeat>
<arch>
<clear_interval>600</clear_interval>
<clear_flag>0</clear_flag>
</arch>
<send>
<ip>192.168.159.131</ip>
<mgr_port>5345</mgr_port>
<data_port>5346</data_port>
<net_pack_size>256</net_pack_size>
<net_turns>0</net_turns>
<crc_check>0</crc_check>
<trigger>0</trigger>
<constraint>0</constraint>
<identity>0</identity>
<filter>
<enable>
<item>APP.</item>
</enable>
</filter>
<map>
<item>APP. == APP.*</item>
</map>
</send>
</cpt>
</dmhs>
配置目的端：
Bash
su - dmdba
cd /dm8/dmhs
复制一份bin目录，单独配置，当同一个服务上需要部署多个dmhs，可以方便区分
cp -r bin cptkafka
cd cptkafka

vim dmhs.hs
内容如下：
<?xml version=“1.0” encoding=“GB2312” standalone=“no”?>
<dmhs>
<base>
<lang>en</lang>
<mgr_port>5345</mgr_port>
<ckpt_interval>60</ckpt_interval>
<siteid>2</siteid>
<version>2.0</version>
</base>
<exec>
<recv>
<data_port>5346</data_port>
</recv>
<exec_thr>1</exec_thr>
<exec_sql>1024</exec_sql>
<exec_policy>2</exec_policy>
<is_kafka>1</is_kafka>
<max_packet_size>16</max_packet_size>
<enable_ddl>1</enable_ddl>
</exec>
</dmhs>

vim dmhs_kafka.properties
内容如下：
##DMHS config file path
dmhs.conf.path=/dm8/dmhs/cptkafka/dmhs.hs
##kafka broker list,such as ip1:port1,ip2:port2,…
bootstrap.servers=192.168.159.131:9092
##kafka topic name
kafka.topic.name=test
##dmhs.sendKey.parse.format=schema:source:tableName
##dmhs.sendKey.parse.format=primary_keys_values
##dmhs.sendTopic.parse.format=schema:source:tableName
##topic.map.conf.path=/dmhs_kafka/bin_0329/tableTopicMap.properties
##whether to enable JSON format check
json.format.check=1
##How many messages print cost time
print.message.num=1000
##How many messages batch to get
dmhs.min.batch.size=100
##kafka serializer class
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
##kafka partitioner config
partitioner.class=com.dameng.dmhs.dmga.service.impl.OnePartitioner
##kafka request acks config
acks=-1
max.request.size=5024000
batch.size=1048576
linger.ms=3
buffer.memory=134217728
retries=3
##enable.idempotence=true
compression.type=none
max.in.flight.requests.per.connection=1
send.buffer.bytes=1048576
metadata.max.age.ms=300000

vim start_dmhs_kafka.sh
#内容如下：
export LANG=en_US.UTF-8
java -Djava.ext.dirs="/usr/local/kafka/libs:." com.dameng.dmhs.dmga.service.impl.ExecDMHSKafkaService dmhs_kafka.properties

chmod +x start_dmhs_kafka.sh

4、同步测试

启动目的端
Shell
su - dmdba
cd /dm8/dmhs/cptkafka
./start_dmhs_kafka.sh

启动后则表示目的端已经以前台方式启动增量同步。
启动源端：
Shell
su - dmdba
cd /dm8/dmhs/cptkafka
./dmhs_server

DMHS>copy 0 “SCH.NAME=‘APP’” DICT
DMHS>copy 0 “SCH.NAME=‘APP’” INSERT
DMHS>clear exec lsn
DMHS>start cpt

查看kafka队列（topic）test中是否有数据
Shell
kafka-console-consumer.sh --bootstrap-server 192.168.159.131:9092 --topic test --from-beginning
配置好服务并启动后，再去源端新增一条数据，然后到目的端kafka看看是否同步成功
Bash

源端添加一条数据

su - dmdba
cd /dm8/dminstall/bin
./disql APP/Dameng123@192.168.159.133:5236
INSERT INTO APP.STUDENTS (NAME, BIRTH_DATE, GENDER, EMAIL, PHONE_NUMBER) VALUES (‘美’, ‘2022-4-25’, ‘F’, ‘meiv@djl.com’, ‘13243253529’);
COMMIT;