1. 首页
  2. 文章中心
  3. C#教程

使用触发器和C#程序实现数据同步

77 阅读 0 评论
我是靠谱客的博主 心灵美诺言，最近开发中收集的这篇文章主要介绍使用触发器和C#程序实现数据同步，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

c#教程icon-default.png?t=M0H8https://www.xin3721.com/eschool/CSharpxin3721/

 

项目中用到了Quartz，也算是Quartz的一个示例。

源库创建数据同步表SYNC_DATA：

CREATE TABLE SYNC_DATA(
    ID                             VARCHAR2(36)      NOT NULL,
    TABLE_NAME                     VARCHAR2(50)      NOT NULL,
    COMMAND                        VARCHAR2(10)      NOT NULL,
    TABLE_KEY                      VARCHAR2(50)      NOT NULL,
    DATA_ID                        VARCHAR2(50)      NOT NULL,
    FAIL_COUNT                     INT               NOT NULL,
    CREATE_TIME                    DATE              NOT NULL,
    UPDATE_TIME                    DATE ,
    CONSTRAINT sync_data_PK PRIMARY KEY (ID)
)
;

COMMENT ON TABLE SYNC_DATA IS '业务数据同步表'
;
COMMENT ON COLUMN SYNC_DATA.ID IS '主键'
;
COMMENT ON COLUMN SYNC_DATA.TABLE_NAME IS '表名'
;
COMMENT ON COLUMN SYNC_DATA.COMMAND IS '命令'
;
COMMENT ON COLUMN SYNC_DATA.TABLE_KEY IS '表主键'
;
COMMENT ON COLUMN SYNC_DATA.DATA_ID IS '数据ID'
;
COMMENT ON COLUMN SYNC_DATA.FAIL_COUNT IS '失败次数'
;
COMMENT ON COLUMN SYNC_DATA.CREATE_TIME IS '创建时间'
;
COMMENT ON COLUMN SYNC_DATA.UPDATE_TIME IS '更新时间'
;

新增修改删除触发器示例(oracle版)：

CREATE OR REPLACE TRIGGER "T_PT_CAMERA_INFO_I"
AFTER INSERT ON "SHINY"."PT_CAMERA_INFO" FOR EACH ROW
BEGIN
    INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
    VALUES(sys_guid(), 'PT_CAMERA_INFO', 'INSERT',  'ID', :new.ID || '', 0,  sysdate);
END;

CREATE OR REPLACE TRIGGER "T_PT_CAMERA_INFO_U"
AFTER UPDATE ON "SHINY"."PT_CAMERA_INFO" FOR EACH ROW
BEGIN
    IF (NVL(to_char(:new.ID), 'X') != NVL(to_char(:old.ID), 'X'))
        OR (NVL(to_char(:new.ASSET_ID), 'X') != NVL(to_char(:old.ASSET_ID), 'X'))
        OR (NVL(to_char(:new.CAMERA_NO), 'X') != NVL(to_char(:old.CAMERA_NO), 'X'))
        OR (NVL(to_char(:new.POSITION_CODE), 'X') != NVL(to_char(:old.POSITION_CODE), 'X'))
        OR (NVL(to_char(:new.POSITION_ID), 'X') != NVL(to_char(:old.POSITION_ID), 'X'))
        OR (NVL(to_char(:new.CAMERA_NAME), 'X') != NVL(to_char(:old.CAMERA_NAME), 'X'))
        OR (NVL(to_char(:new.ORG_ID), 'X') != NVL(to_char(:old.ORG_ID), 'X'))
        OR (NVL(to_char(:new.ADDRESS), 'X') != NVL(to_char(:old.ADDRESS), 'X'))
        OR (NVL(to_char(:new.CAMERA_IP), 'X') != NVL(to_char(:old.CAMERA_IP), 'X'))
        OR (NVL(to_char(:new.LONGITUDE), 'X') != NVL(to_char(:old.LONGITUDE), 'X'))
        OR (NVL(to_char(:new.LATITUDE), 'X') != NVL(to_char(:old.LATITUDE), 'X'))
        OR (NVL(to_char(:new.SN), 'X') != NVL(to_char(:old.SN), 'X'))
        OR (NVL(to_char(:new.CAMERA_MODEL), 'X') != NVL(to_char(:old.CAMERA_MODEL), 'X'))
        OR (NVL(to_char(:new.MANUFACTURER), 'X') != NVL(to_char(:old.MANUFACTURER), 'X'))
        OR (NVL(to_char(:new.SUPPLIER), 'X') != NVL(to_char(:old.SUPPLIER), 'X'))
        OR (NVL(to_char(:new.ADD_ID), 'X') != NVL(to_char(:old.ADD_ID), 'X'))
        OR (NVL(to_char(:new.ADD_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X') != NVL(to_char(:old.ADD_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X'))
        OR (NVL(to_char(:new.MODIFY_ID), 'X') != NVL(to_char(:old.MODIFY_ID), 'X'))
        OR (NVL(to_char(:new.MODIFY_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X') != NVL(to_char(:old.MODIFY_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X'))
        OR (NVL(to_char(:new.USER_NAME), 'X') != NVL(to_char(:old.USER_NAME), 'X'))
        OR (NVL(to_char(:new.PASS_WORD), 'X') != NVL(to_char(:old.PASS_WORD), 'X'))
        OR (NVL(to_char(:new.CASE_INSTALL_TYPE), 'X') != NVL(to_char(:old.CASE_INSTALL_TYPE), 'X'))
        OR (NVL(to_char(:new.AREA_CODE), 'X') != NVL(to_char(:old.AREA_CODE), 'X'))
        OR (NVL(to_char(:new.MONITOR_TYPE), 'X') != NVL(to_char(:old.MONITOR_TYPE), 'X'))
        OR (NVL(to_char(:new.POINT_NAME), 'X') != NVL(to_char(:old.POINT_NAME), 'X'))
        OR (NVL(to_char(:new.CAMERA_IP6), 'X') != NVL(to_char(:old.CAMERA_IP6), 'X'))
        OR (NVL(to_char(:new.SUBNET_MASK), 'X') != NVL(to_char(:old.SUBNET_MASK), 'X'))
        OR (NVL(to_char(:new.GATEWAY), 'X') != NVL(to_char(:old.GATEWAY), 'X'))
        OR (NVL(to_char(:new.MAC_ADDRESS), 'X') != NVL(to_char(:old.MAC_ADDRESS), 'X'))
        OR (NVL(to_char(:new.ONU_SN), 'X') != NVL(to_char(:old.ONU_SN), 'X'))
        OR (NVL(to_char(:new.CAMERA_TYPE), 'X') != NVL(to_char(:old.CAMERA_TYPE), 'X'))
        OR (NVL(to_char(:new.CAMERA_FUN_TYPE), 'X') != NVL(to_char(:old.CAMERA_FUN_TYPE), 'X'))
        OR (NVL(to_char(:new.FILL_LIGTH_ATTR), 'X') != NVL(to_char(:old.FILL_LIGTH_ATTR), 'X'))
        OR (NVL(to_char(:new.CAMERA_ENCODE_TYPE), 'X') != NVL(to_char(:old.CAMERA_ENCODE_TYPE), 'X'))
        OR (NVL(to_char(:new.POWER_TAKE_TYPE), 'X') != NVL(to_char(:old.POWER_TAKE_TYPE), 'X'))
        OR (NVL(to_char(:new.POWER_TAKE_LENGTH), 'X') != NVL(to_char(:old.POWER_TAKE_LENGTH), 'X'))
        OR (NVL(to_char(:new.SOUND_ALARM), 'X') != NVL(to_char(:old.SOUND_ALARM), 'X'))
        OR (NVL(to_char(:new.RESOLUTION), 'X') != NVL(to_char(:old.RESOLUTION), 'X'))
        OR (NVL(to_char(:new.SOFT_VERSION), 'X') != NVL(to_char(:old.SOFT_VERSION), 'X'))
        OR (NVL(to_char(:new.LENS_PARAM), 'X') != NVL(to_char(:old.LENS_PARAM), 'X'))
        OR (NVL(to_char(:new.IS_HAVE_CONSOLE), 'X') != NVL(to_char(:old.IS_HAVE_CONSOLE), 'X'))
        OR (NVL(to_char(:new.INSTALL_WAY), 'X') != NVL(to_char(:old.INSTALL_WAY), 'X'))
        OR (NVL(to_char(:new.LINEAR_WAY), 'X') != NVL(to_char(:old.LINEAR_WAY), 'X'))
        OR (NVL(to_char(:new.RESOURCE_PLACE), 'X') != NVL(to_char(:old.RESOURCE_PLACE), 'X'))
        OR (NVL(to_char(:new.IMPORT_WATCH), 'X') != NVL(to_char(:old.IMPORT_WATCH), 'X'))
        OR (NVL(to_char(:new.POSITION_TYPE), 'X') != NVL(to_char(:old.POSITION_TYPE), 'X'))
        OR (NVL(to_char(:new.COMMUNITY), 'X') != NVL(to_char(:old.COMMUNITY), 'X'))
        OR (NVL(to_char(:new.STREET), 'X') != NVL(to_char(:old.STREET), 'X'))
        OR (NVL(to_char(:new.WATCH_SPEC_LOCATION), 'X') != NVL(to_char(:old.WATCH_SPEC_LOCATION), 'X'))
        OR (NVL(to_char(:new.ROAD_DIRECTION), 'X') != NVL(to_char(:old.ROAD_DIRECTION), 'X'))
        OR (NVL(to_char(:new.FOUL_LINE), 'X') != NVL(to_char(:old.FOUL_LINE), 'X'))
        OR (NVL(to_char(:new.FEN_JU), 'X') != NVL(to_char(:old.FEN_JU), 'X'))
        OR (NVL(to_char(:new.POLICE_STATION), 'X') != NVL(to_char(:old.POLICE_STATION), 'X'))
        OR (NVL(to_char(:new.CAMERA_DIRECTION), 'X') != NVL(to_char(:old.CAMERA_DIRECTION), 'X'))
        OR (NVL(to_char(:new.INSTALL_HEIGHT), 'X') != NVL(to_char(:old.INSTALL_HEIGHT), 'X'))
        OR (NVL(to_char(:new.CROSS_ARM1), 'X') != NVL(to_char(:old.CROSS_ARM1), 'X'))
        OR (NVL(to_char(:new.CROSS_ARM2), 'X') != NVL(to_char(:old.CROSS_ARM2), 'X'))
        OR (NVL(to_char(:new.INDOOR_OR_NOT), 'X') != NVL(to_char(:old.INDOOR_OR_NOT), 'X'))
        OR (NVL(to_char(:new.SPECIAL_PHOTO_PATH), 'X') != NVL(to_char(:old.SPECIAL_PHOTO_PATH), 'X'))
        OR (NVL(to_char(:new.LOCATION_PHOTO_PATH), 'X') != NVL(to_char(:old.LOCATION_PHOTO_PATH), 'X'))
        OR (NVL(to_char(:new.REAL_PHOTO_PATH), 'X') != NVL(to_char(:old.REAL_PHOTO_PATH), 'X'))
        OR (NVL(to_char(:new.NETWORK_PROPERTIES), 'X') != NVL(to_char(:old.NETWORK_PROPERTIES), 'X'))
        OR (NVL(to_char(:new.POLICE_AREA_CODE), 'X') != NVL(to_char(:old.POLICE_AREA_CODE), 'X'))
        OR (NVL(to_char(:new.INSTALL_PERSION), 'X') != NVL(to_char(:old.INSTALL_PERSION), 'X'))
        OR (NVL(to_char(:new.INSTALL_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X') != NVL(to_char(:old.INSTALL_TIME, 'yyyy-mm-dd hh24:mi:ss'),'X'))
        OR (NVL(to_char(:new.BUILD_PERIOD), 'X') != NVL(to_char(:old.BUILD_PERIOD), 'X'))
        OR (NVL(to_char(:new.PROJECT_NAME), 'X') != NVL(to_char(:old.PROJECT_NAME), 'X'))
        OR (NVL(to_char(:new.MANAGER_UNIT), 'X') != NVL(to_char(:old.MANAGER_UNIT), 'X'))
        OR (NVL(to_char(:new.MANAGER_UNIT_TEL), 'X') != NVL(to_char(:old.MANAGER_UNIT_TEL), 'X'))
        OR (NVL(to_char(:new.MAINTAIN_UNIT), 'X') != NVL(to_char(:old.MAINTAIN_UNIT), 'X'))
        OR (NVL(to_char(:new.MAINTAIN_UNIT_TEL), 'X') != NVL(to_char(:old.MAINTAIN_UNIT_TEL), 'X'))
        OR (NVL(to_char(:new.RECODE_SAVE_TYPE), 'X') != NVL(to_char(:old.RECODE_SAVE_TYPE), 'X'))
        OR (NVL(to_char(:new.DEVICE_STATE), 'X') != NVL(to_char(:old.DEVICE_STATE), 'X'))
        OR (NVL(to_char(:new.INDUSTRY_OWN), 'X') != NVL(to_char(:old.INDUSTRY_OWN), 'X'))
        OR (NVL(to_char(:new.IS_REGISTER_IMOS), 'X') != NVL(to_char(:old.IS_REGISTER_IMOS), 'X'))
        OR (NVL(to_char(:new.IS_WIFI), 'X') != NVL(to_char(:old.IS_WIFI), 'X'))
        OR (NVL(to_char(:new.IS_FLASH), 'X') != NVL(to_char(:old.IS_FLASH), 'X'))
        OR (NVL(to_char(:new.CAMERA_NO_STR), 'X') != NVL(to_char(:old.CAMERA_NO_STR), 'X'))
        OR (NVL(to_char(:new.CAMERA_VCN_CODE), 'X') != NVL(to_char(:old.CAMERA_VCN_CODE), 'X'))
        OR (NVL(to_char(:new.FIELD_NO), 'X') != NVL(to_char(:old.FIELD_NO), 'X'))
        OR (NVL(to_char(:new.KEY_UNIT), 'X') != NVL(to_char(:old.KEY_UNIT), 'X'))
        OR (NVL(to_char(:new.UNIT_TYPE), 'X') != NVL(to_char(:old.UNIT_TYPE), 'X'))
        OR (NVL(to_char(:new.SHOW_LEVEL), 'X') != NVL(to_char(:old.SHOW_LEVEL), 'X'))
        OR (NVL(to_char(:new.PROTOCOL_TYPE), 'X') != NVL(to_char(:old.PROTOCOL_TYPE), 'X'))
        OR (NVL(to_char(:new.CAMERA_PORT), 'X') != NVL(to_char(:old.CAMERA_PORT), 'X'))
        OR (NVL(to_char(:new.INTERFACE_TYPE), 'X') != NVL(to_char(:old.INTERFACE_TYPE), 'X'))
        OR (NVL(to_char(:new.CHANNEL), 'X') != NVL(to_char(:old.CHANNEL), 'X'))
        OR (NVL(to_char(:new.USER_OBJECT), 'X') != NVL(to_char(:old.USER_OBJECT), 'X'))
        OR (NVL(to_char(:new.IMG_PATH), 'X') != NVL(to_char(:old.IMG_PATH), 'X'))
        OR (NVL(to_char(:new.CAMERA_DESC), 'X') != NVL(to_char(:old.CAMERA_DESC), 'X'))
        OR (NVL(to_char(:new.IS_REGISTER_VCN), 'X') != NVL(to_char(:old.IS_REGISTER_VCN), 'X'))
        OR (NVL(to_char(:new.IS_DEL), 'X') != NVL(to_char(:old.IS_DEL), 'X'))
        OR (NVL(to_char(:new.ORDER_VALUE), 'X') != NVL(to_char(:old.ORDER_VALUE), 'X'))
        OR (NVL(to_char(:new.POLLING_RESULT), 'X') != NVL(to_char(:old.POLLING_RESULT), 'X'))
        OR (NVL(to_char(:new.SERVER_ID), 'X') != NVL(to_char(:old.SERVER_ID), 'X'))
        OR (NVL(to_char(:new.SHORT_MSG), 'X') != NVL(to_char(:old.SHORT_MSG), 'X'))
        OR (NVL(to_char(:new.CAMERA_BELONGS_ID), 'X') != NVL(to_char(:old.CAMERA_BELONGS_ID), 'X'))
        OR (NVL(to_char(:new.RELATED_CUSTOMS), 'X') != NVL(to_char(:old.RELATED_CUSTOMS), 'X'))
        OR (NVL(to_char(:new.ADDED_TO_SDE), 'X') != NVL(to_char(:old.ADDED_TO_SDE), 'X'))
        OR (NVL(to_char(:new.CAMERA_BAK), 'X') != NVL(to_char(:old.CAMERA_BAK), 'X'))
        OR (NVL(to_char(:new.CAMERA_BELONGS_PK), 'X') != NVL(to_char(:old.CAMERA_BELONGS_PK), 'X'))
        OR (NVL(to_char(:new.MEMBERBAR_CODE), 'X') != NVL(to_char(:old.MEMBERBAR_CODE), 'X'))
        OR (NVL(to_char(:new.IS_BRANCH), 'X') != NVL(to_char(:old.IS_BRANCH), 'X'))
        OR (NVL(to_char(:new.IS_WATCHPOS), 'X') != NVL(to_char(:old.IS_WATCHPOS), 'X'))
        OR (NVL(to_char(:new.CAMERA_ANGLE), 'X') != NVL(to_char(:old.CAMERA_ANGLE), 'X'))
        OR (NVL(to_char(:new.CROSS_ARM3), 'X') != NVL(to_char(:old.CROSS_ARM3), 'X'))
        OR (NVL(to_char(:new.IS_SYS), 'X') != NVL(to_char(:old.IS_SYS), 'X'))
        OR (NVL(to_char(:new.RECORD_TIME), 'X') != NVL(to_char(:old.RECORD_TIME), 'X'))
        OR (NVL(to_char(:new.ANALYSIS_NO), 'X') != NVL(to_char(:old.ANALYSIS_NO), 'X'))
        OR (NVL(to_char(:new.WIFI_STATE), 'X') != NVL(to_char(:old.WIFI_STATE), 'X'))
        OR (NVL(to_char(:new.FACE_TASK_STATUS), 'X') != NVL(to_char(:old.FACE_TASK_STATUS), 'X'))
        OR (NVL(to_char(:new.VIDEO_TASK_STATUS), 'X') != NVL(to_char(:old.VIDEO_TASK_STATUS), 'X'))
        OR (NVL(to_char(:new.BAYONET_TASK_STATUS), 'X') != NVL(to_char(:old.BAYONET_TASK_STATUS), 'X'))
        OR (NVL(to_char(:new.VQD_URL), 'X') != NVL(to_char(:old.VQD_URL), 'X'))
        OR (NVL(to_char(:new.SYS_TYPE), 'X') != NVL(to_char(:old.SYS_TYPE), 'X'))
        OR (NVL(to_char(:new.IS_HAVE_CAPTURE), 'X') != NVL(to_char(:old.IS_HAVE_CAPTURE), 'X'))

    THEN
        INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
        VALUES(sys_guid(), 'PT_CAMERA_INFO', 'UPDATE',  'ID', :new.ID || '', 0,  sysdate);
    END IF;
END;

CREATE OR REPLACE TRIGGER "T_PT_CAMERA_INFO_D"
BEFORE DELETE ON "SHINY"."PT_CAMERA_INFO" FOR EACH ROW
BEGIN
    INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
    VALUES(sys_guid(), 'PT_CAMERA_INFO', 'DELETE',  'ID', :old.ID || '', 0,  sysdate);
END;

CREATE OR REPLACE TRIGGER "T_VIDEO_VIDEOSOURCE_I"
AFTER INSERT ON "SHINY"."VIPF_VIDEO_VIDEOSOURCE" FOR EACH ROW
BEGIN
    INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
    VALUES(sys_guid(), 'VIPF_VIDEO_VIDEOSOURCE', 'INSERT',  'CAMERA_ID', :new.CAMERA_ID || '', 0,  sysdate);
END;

CREATE OR REPLACE TRIGGER "T_VIDEO_VIDEOSOURCE_U"
AFTER UPDATE ON "SHINY"."VIPF_VIDEO_VIDEOSOURCE" FOR EACH ROW
BEGIN
    INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
    VALUES(sys_guid(), 'VIPF_VIDEO_VIDEOSOURCE', 'UPDATE',  'CAMERA_ID', :new.CAMERA_ID || '', 0,  sysdate);
END;

CREATE OR REPLACE TRIGGER "T_VIDEO_VIDEOSOURCE_D"
BEFORE DELETE ON "SHINY"."VIPF_VIDEO_VIDEOSOURCE" FOR EACH ROW
BEGIN
    INSERT INTO sync_data(ID, TABLE_NAME, COMMAND, TABLE_KEY, DATA_ID, FAIL_COUNT, CREATE_TIME)
    VALUES(sys_guid(), 'VIPF_VIDEO_VIDEOSOURCE', 'DELETE',  'CAMERA_ID', :old.CAMERA_ID || '', 0,  sysdate);
END;

启动数据同步任务C#代码：

using DataSync.DAL;
using DataSync.Utils;
using Utils;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Configuration;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.ServiceProcess;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Timers;
using System.Windows.Forms;
using Quartz.Impl;
using Quartz;
using DataSync.Jobs;
using Models;

namespace DataSync
{
    /// <summary>
    /// 数据同步服务
    /// by suxiang
    /// </summary>
    partial class DataSyncService : ServiceBase
    {
        #region 变量

        #endregion

        #region 构造函数
        public DataSyncService()
        {
            InitializeComponent();
        }
        #endregion

        #region OnStart 启动服务
        protected override void OnStart(string[] args)
        {
            startScheduler();

            LogUtil.Log("服务启动成功");
        }
        #endregion

        #region OnStop 停止服务
        protected override void OnStop()
        {
            LogUtil.Log("服务停止成功");

            Thread.Sleep(100); //等待一会，待日志写入文件
        }
        #endregion

        #region Start 启动服务
        public void Start()
        {
            OnStart(null);
        }
        #endregion

        #region scheduleJob
        private async void startScheduler()
        {
            TaskSchedulerEx taskEx = new TaskSchedulerEx(8, 64);
            StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
            IScheduler scheduler = await schedulerFactory.GetScheduler();
            await scheduler.Start();
            await LogUtil.Log("任务调度器已启动");

            if (ConfigUtil.SyncParam == 1) //部署在政务网侧 
            {
                scheduleJob(scheduler, "政务网中间库->政务网平台库", ConfigUtil.ZWW_Middle, ConfigUtil.ZWW_PT, taskEx);
                scheduleJob(scheduler, "政务网平台库->政务网中间库", ConfigUtil.ZWW_PT, ConfigUtil.ZWW_Middle, taskEx);
            }

            if (ConfigUtil.SyncParam == 2) //部署在视频网侧
            {
                scheduleJob(scheduler, "视频网平台库->视频网中间库", ConfigUtil.SPW_PT, ConfigUtil.SPW_Middle, taskEx);
            }
        }

        private async void scheduleJob(IScheduler scheduler, string taskName, string dbSourceConnectionString, string dbTargetConnectionString, TaskSchedulerEx taskEx)
        {
            string cronString = "0/5 * * * * ?";
            IJobDetail jobDetail = JobBuilder.Create<SyncJob>().Build();
            ITrigger trigger = TriggerBuilder.Create().WithCronSchedule(cronString).Build();
            SpeedInfo speedInfo = new SpeedInfo();
            jobDetail.JobDataMap.Put("taskName", taskName);
            jobDetail.JobDataMap.Put("dbSourceConnectionString", dbSourceConnectionString);
            jobDetail.JobDataMap.Put("dbTargetConnectionString", dbTargetConnectionString);
            jobDetail.JobDataMap.Put("speedInfo", speedInfo);
            jobDetail.JobDataMap.Put("taskEx", taskEx);
            await scheduler.ScheduleJob(jobDetail, trigger);

            await scheduleCalcSpeedJob(scheduler, taskName, speedInfo);
        }

        private async Task scheduleCalcSpeedJob(IScheduler scheduler, string taskName, SpeedInfo speedInfo)
        {
            int timeRange = 10;
            string cronString = "0/" + timeRange + " * * * * ?";
            IJobDetail jobDetail = JobBuilder.Create<CalcSpeedJob>().Build();
            ITrigger trigger = TriggerBuilder.Create().WithCronSchedule(cronString).Build();
            jobDetail.JobDataMap.Put("taskName", taskName);
            jobDetail.JobDataMap.Put("timeRange", timeRange);
            jobDetail.JobDataMap.Put("speedInfo", speedInfo);
            await scheduler.ScheduleJob(jobDetail, trigger);
        }
        #endregion

    }
}

数据同步任务C#代码：

using DataSync.DAL;
using DataSync.Models;
using DataSync.Utils;
using Models;
using Oracle.ManagedDataAccess.Client;
using Quartz;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Utils;

namespace DataSync.Jobs
{
    /// <summary>
    /// 数据同步任务
    /// </summary>
    [DisallowConcurrentExecution]
    public class SyncJob : IJob
    {
        #region 变量属性
        private readonly string INSERT = "INSERT";
        private readonly string UPDATE = "UPDATE";
        private readonly string DELETE = "DELETE";
        #endregion

        #region 数据库操作类
        private OracleHelper _dbSource = null;
        private OracleHelper _dbTarget = null;
        #endregion

        #region 任务参数
        private string _taskName = null;
        private string _dbSourceConnectionString = null;
        private string _dbTargetConnectionString = null;
        private SpeedInfo _speedInfo = null;
        private TaskSchedulerEx _task = null;
        #endregion

        public Task Execute(IJobExecutionContext context)
        {
            return Task.Run(() =>
            {
                try
                {
                    _taskName = context.JobDetail.JobDataMap["taskName"].ToString();
                    _dbSourceConnectionString = context.JobDetail.JobDataMap["dbSourceConnectionString"].ToString();
                    _dbTargetConnectionString = context.JobDetail.JobDataMap["dbTargetConnectionString"].ToString();
                    _speedInfo = (SpeedInfo)context.JobDetail.JobDataMap["speedInfo"];
                    _task = (TaskSchedulerEx)context.JobDetail.JobDataMap["taskEx"];

                    _dbSource = new OracleHelper(_dbSourceConnectionString);
                    _dbTarget = new OracleHelper(_dbTargetConnectionString);

                    List<SYNC_DATA> list = ServiceHelper.Get<SyncDataDal>().GetList(_dbSourceConnectionString);
                    int oldCount = list.Count;
                    list = DictinctSyncData(list);
                    if (oldCount > 0 && oldCount != list.Count) LogUtil.Log(string.Format("SYNC_DATA集合去重前数量：{0} 去重后数量：{1}", oldCount, list.Count));
                    Dictionary<string, List<SYNC_DATA>> dict = list.ToLookup(a => a.TABLE_NAME).ToDictionary(a => a.Key, a => a.ToList()); //按表分组，对于单张表，必须用单线程，保证处理顺序

                    int syncCount = 0;
                    object lockObj = new object();
                    List<Task> taskList = new List<Task>();
                    foreach (string key in dict.Keys)
                    {
                        foreach (SYNC_DATA item in dict[key])
                        {
                            Task task = _task.Run((obj) =>
                            {
                                SYNC_DATA syncData = obj as SYNC_DATA;
                                string targetTableName = syncData.TABLE_NAME;

                                //表名不一至的特殊情况
                                if (ConfigUtil.SyncParam == 1) //部署在政务网侧 
                                {
                                    if (syncData.TABLE_NAME == "VIPF_VIDEO_VIDEOSOURCE") targetTableName = "ZZ_VIDEO_VIDEOSOURCE1";
                                }

                                //查询并缓存目标数据库表字段信息集合
                                List<DbColumnInfo> columnList = MemoryCacheUtil.TryGetValue<List<DbColumnInfo>>(_dbSourceConnectionString + targetTableName, () => _dbTarget.GetAllColumns(targetTableName));

                                int row;
                                SyncOneData(_dbSourceConnectionString, (SYNC_DATA)syncData, targetTableName, _dbSource, _dbTarget, columnList, out row);
                                lock (lockObj) { syncCount += row; }

                                _speedInfo.addCount();
                            }, item);
                            taskList.Add(task);
                        }
                    }
                    Task.WaitAll(taskList.ToArray());
                    if (syncCount > 0) LogUtil.Log(string.Format("本次成功同步 {0} 条数据", syncCount));
                }
                catch (Exception ex)
                {
                    LogUtil.Error(ex, "数据同步任务 出错");
                }
            });
        }

        #region 同步一条数据
        /// <summary>
        /// 同步一条数据
        /// </summary>
        private void SyncOneData(string connStr, SYNC_DATA item, string targetTableName, OracleHelper dbSource, OracleHelper dbTarget, List<DbColumnInfo> columnList, out int row)
        {
            bool isFault = false;
            DateTime dt = DateTime.Now;
            row = 0;

            try
            {
                #region 通过表名查询字段集合
                string columns = string.Join(",", columnList.ConvertAll<string>(a => a.columns_name));
                string columnsForInsert = string.Join(",", columnList.ConvertAll<string>(a => ":" + a.columns_name));
                string columnsForUpdate = string.Join(",", columnList.ConvertAll<string>(a => a.columns_name + "=:" + a.columns_name));
                #endregion

                if (item.COMMAND == INSERT || item.COMMAND == UPDATE)
                {
                    //数据查询
                    DataTable sourceData = dbSource.Query(string.Format(@"select * from {0} where {1}", item.TABLE_NAME, CreateWhere(item.TABLE_KEY, item.DATA_ID)));
                    DataTable targetData = dbTarget.Query(string.Format(@"select * from {0} where {1}", targetTableName, CreateWhere(item.TABLE_KEY, item.DATA_ID)));

                    //判断摄像机表是否需要同步
                    if (item.TABLE_NAME == "PT_CAMERA_INFO")
                    {
                        if (!needSync(sourceData, targetData))
                        {
                            ServiceHelper.Get<SyncDataDal>().DeleteById(connStr, item.ID);
                            return;
                        }
                    }

                    if (sourceData != null && sourceData.Rows.Count > 0)
                    {
                        bool exists = targetData != null && targetData.Rows.Count > 0;

                        if ((item.COMMAND == INSERT && !exists) || (item.COMMAND == UPDATE && !exists))
                        {
                            #region 插入数据
                            string insertSql = string.Format("insert into {0}({1}) values({2})", targetTableName, columns, columnsForInsert);

                            #region 字段赋值
                            OracleParameter[] paramArr = new OracleParameter[columnList.Count];

                            for (int i = 0; i < columnList.Count; i++)
                            {
                                DbColumnInfo columnInfo = columnList[i];

                                paramArr[i] = new OracleParameter(columnInfo.columns_name, GetColumnData(sourceData, columnInfo.columns_name));
                            }
                            #endregion

                            row = dbTarget.ExecuteSql(insertSql, paramArr);

                            if (row > 0) LogUtil.Log(string.Format("{0} 表 {1} 插入一条数据，{2}", _taskName, targetTableName, CreateWhereLog(item.TABLE_KEY, item.DATA_ID)));
                            #endregion
                        }

                        if ((item.COMMAND == INSERT && exists) || (item.COMMAND == UPDATE && exists))
                        {
                            #region 更新数据
                            string updateSql = string.Format("update {0} set {1} where {2}", targetTableName, columnsForUpdate, CreateWhere(item.TABLE_KEY, item.DATA_ID));

                            #region 字段赋值
                            List<OracleParameter> paramList = new List<OracleParameter>();

                            foreach (DbColumnInfo columnInfo in columnList)
                            {
                                paramList.Add(new OracleParameter(columnInfo.columns_name, GetColumnData(sourceData, columnInfo.columns_name)));
                            }
                            #endregion

                            row = dbTarget.ExecuteSql(updateSql, paramList.ToArray());

                            if (row > 0) LogUtil.Log(string.Format("{0} 表 {1} 更新一条数据，{2}", _taskName, targetTableName, CreateWhereLog(item.TABLE_KEY, item.DATA_ID)));
                            #endregion
                        }
                    }
                }

                if (item.COMMAND == DELETE)
                {
                    #region 删除数据
                    string deleteSql = string.Format("delete from {0} where {1}", targetTableName, CreateWhere(item.TABLE_KEY, item.DATA_ID));

                    row = dbTarget.ExecuteSql(deleteSql);

                    if (row > 0) LogUtil.Log(string.Format("{0} 表 {1} 删除一条数据，{2}", _taskName, targetTableName, CreateWhereLog(item.TABLE_KEY, item.DATA_ID)));
                    #endregion
                }

                ServiceHelper.Get<SyncDataDal>().DeleteById(connStr, item.ID);

            }
            catch (Exception ex)
            {
                isFault = true;
                LogUtil.Error(ex, "同步数据失败，表名：" + item.TABLE_NAME.ToUpper() + " 命令：" + item.COMMAND + " 数据ID：" + item.DATA_ID);
            }

            try
            {
                if (isFault)
                {
                    SYNC_DATA sysSyncBs = ServiceHelper.Get<SyncDataDal>().GetById(connStr, item.ID);
                    sysSyncBs.FAIL_COUNT++;
                    ServiceHelper.Get<SyncDataDal>().Update(connStr, sysSyncBs);
                    LogUtil.Debug("更新业务数据同步表SYNC_DATA，ID：" + item.ID);
                }
            }
            catch (Exception ex)
            {
                LogUtil.Error(ex, "更新业务数据同步表SYNC_DATA出错");
            }
        }
        #endregion

        #region 判断是否是主键
        /// <summary>
        /// 判断是否是主键
        /// </summary>
        private bool IsIdField(string columnName, string idField)
        {
            if (idField.IndexOf(',') >= 0)
            {
                string[] idFieldArr = idField.Split(',');
                return idFieldArr.ToList().Exists(a => a == columnName);
            }
            else
            {
                return columnName == idField;
            }
        }
        #endregion

        #region 根据主键生成Where条件语句
        /// <summary>
        /// 根据主键生成Where条件语句
        /// </summary>
        private string CreateWhere(string idField, string idData)
        {
            if (idField.IndexOf(',') >= 0)
            {
                StringBuilder sb = new StringBuilder(" 1=1 ");
                string[] idFieldArr = idField.Split(',');
                string[] idDataArr = idData.Split(',');
                for (int i = 0; i < idFieldArr.Length; i++)
                {
                    sb.AppendFormat(" and {0}='{1}' ", idFieldArr[i], idDataArr[i]);
                }
                return sb.ToString();
            }
            else
            {
                return string.Format("{0}='{1}'", idField, idData);
            }
        }
        #endregion

        #region 根据字段名从DataTable中获取数据
        /// <summary>
        /// 根据字段名从DataTable中获取数据
        /// </summary>
        private object GetColumnData(DataTable data, string columnName)
        {
            for (int i = 0; i < data.Columns.Count; i++)
            {
                if (data.Columns[i].ColumnName == columnName)
                {
                    return data.Rows[0][data.Columns[i]];
                }
            }
            return DBNull.Value;
        }
        #endregion

        #region 根据主键生成Where条件语句日志信息
        /// <summary>
        /// 根据主键生成Where条件语句日志信息
        /// </summary>
        private string CreateWhereLog(string idField, string idData)
        {
            if (idField.IndexOf(',') >= 0)
            {
                List<string> list = new List<string>();
                string[] idFieldArr = idField.Split(',');
                string[] idDataArr = idData.Split(',');
                for (int i = 0; i < idFieldArr.Length; i++)
                {
                    list.Add(string.Format("{0}:{1}", idFieldArr[i], idDataArr[i]));
                }
                return string.Join("，", list.ToArray());
            }
            else
            {
                return string.Format("{0}:{1}", idField, idData);
            }
        }
        #endregion

        #region 判断是否需要同步
        private bool needSync(DataTable sourceTable, DataTable targetTable)
        {
            HashSet<string> hashSet = new HashSet<string>() { "POLLING_TIME", "CAMERA_STATE" };

            if (targetTable == null || targetTable.Rows.Count == 0) return true;

            if (sourceTable != null && sourceTable.Rows.Count > 0 && targetTable != null && targetTable.Rows.Count > 0)
            {
                foreach (DataColumn column in sourceTable.Columns)
                {
                    if (targetTable.Columns.Contains(column.ColumnName))
                    {
                        if (sourceTable.Rows[0][column.ColumnName] != null && !sourceTable.Rows[0][column].Equals(targetTable.Rows[0][column.ColumnName]))
                        {
                            if (!hashSet.Contains(column.ColumnName))
                            {
                                return true;
                            }
                        }
                    }
                }
            }
            return false;
        }
        #endregion

        #region SYNC_DATA集合去重,对于按条件相同的数据,按顺序保留最后一条
        /// <summary>
        /// SYNC_DATA集合去重,对于按条件相同的数据,按顺序保留最后一条
        /// </summary>
        private List<SYNC_DATA> DictinctSyncData(List<SYNC_DATA> list)
        {
            List<SYNC_DATA> result = new List<SYNC_DATA>();

            for (int i = 0; i < list.Count; i++)
            {
                SYNC_DATA dataI = list[i];

                bool exists = false;
                for (int j = i + 1; j < list.Count; j++)
                {
                    SYNC_DATA dataJ = list[j];

                    if (dataI.TABLE_NAME == dataJ.TABLE_NAME
                        && dataI.TABLE_KEY == dataJ.TABLE_KEY
                        && dataI.DATA_ID == dataJ.DATA_ID)
                    {
                        exists = true;
                    }
                }

                if (!exists)
                {
                    result.Add(dataI);
                }
            }

            return result;
        }
        #endregion

    }

}

 

最后

以上就是心灵美诺言为你收集整理的使用触发器和C#程序实现数据同步的全部内容，希望文章能够帮你解决使用触发器和C#程序实现数据同步所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
点赞(51)

相关文章

JS模拟键盘输入及手动触发事件的方式
JS模拟键盘输入及手动触发事件的方式
oracle实训七作业代码,【雪野实训记录】Oracle数据库 T8作业——触发器
oracle实训七作业代码,【雪野实训记录】Oracle数据库 T8作业——触发器
PLSQL编写触发器
PLSQL编写触发器
html模拟点击某个键盘按钮,如何使用JavaScript模拟按键或单击？
html模拟点击某个键盘按钮,如何使用JavaScript模拟按键或单击？
使用触发器和C#程序实现数据同步
使用触发器和C#程序实现数据同步
mysql触发器语法错误_mysql触发器语法错误问题
mysql触发器语法错误_mysql触发器语法错误问题
unity如何取消鼠标与GameObject的trigger的触发，OnMouseEnter不起作用
unity如何取消鼠标与GameObject的trigger的触发，OnMouseEnter不起作用
oracle基于视图的instead of trigger丢失
oracle基于视图的instead of trigger丢失

评论列表共有 0 条评论

立即
投稿
返回
顶部