using Newtonsoft.Json;using StackExchange.Redis;using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Data;using System.Data.SqlClient;using System.Linq;using System.Threading.Tasks;namespace WorkSample.StackExchange.Redis{ class Program { // static readonly ConcurrentQueue vehicleStationStatementsCenter = new ConcurrentQueue (); static void Main(string[] args) { // 1 001 广州东东信息技术有限公司 // 2 002 广州西西信息技术有限公司 var settlementCompanies = new List { new SettlementCompany { Id = 1, Code = "001", Name = "广州东东信息技术有限公司" }, new SettlementCompany { Id = 2, Code = "002", Name = "广州西西信息技术有限公司" } }; // 汽车站场信息 // 1 001 东站 conn001token12 // 2 002 西站 conn002token12 var vehicleStations = new List { new VehicleStation { Id = 1, Code = "001", Name = "东站", Token = @" Data Source = (localdb)\MSSQLLocalDB; Initial Catalog = VehicleStationEast; Integrated Security = True; Connect Timeout = 30; Encrypt = False; TrustServerCertificate = False; ApplicationIntent = ReadWrite; MultiSubnetFailover = False" }, new VehicleStation { Id = 2, Code = "002", Name = "西站", Token = @" Data Source = (localdb)\MSSQLLocalDB; Initial Catalog = VehicleStationWest; Integrated Security = True; Connect Timeout = 30; Encrypt = False; TrustServerCertificate = False; ApplicationIntent = ReadWrite; MultiSubnetFailover = False" } }; // 校准配置信息 // 1 001 东站 001 广州东东信息技术有限公司 00010 东东 // 2 001 东站 002 广州西西信息技术有限公司 00012 西西 // 3 002 西站 001 广州东东信息技术有限公司 00001 广州东东信息技术有限公司 // 4 002 西站 002 广州西西信息技术有限公司 00002 广州西西信息技术有限公司 var correctionConfigurations = new List { new CorrectionConfiguration { Id = 1, VehicleStationCode = "001", VehicleStationName = "东站", SettlementCompanyCode = "001", SettlementCompanyName = "广州东东信息技术有限公司", CorrectionSettlementCompanyCode = "00010", CorrectionSettlementCompanyName = "东东" }, new CorrectionConfiguration { Id = 2, VehicleStationCode = "001", VehicleStationName = "东站", SettlementCompanyCode = "002", SettlementCompanyName = "广州西西信息技术有限公司", CorrectionSettlementCompanyCode = "00012", CorrectionSettlementCompanyName = "西西" }, new CorrectionConfiguration { Id = 3, VehicleStationCode = "002", VehicleStationName = "西站", SettlementCompanyCode = "001", SettlementCompanyName = "广州东东信息技术有限公司", CorrectionSettlementCompanyCode = "00001", CorrectionSettlementCompanyName = "广州东东信息技术有限公司" }, new CorrectionConfiguration { Id = 4, VehicleStationCode = "002", VehicleStationName = "西站", SettlementCompanyCode = "002", SettlementCompanyName = "广州西西信息技术有限公司", CorrectionSettlementCompanyCode = "00002", CorrectionSettlementCompanyName = "广州西西信息技术有限公司" }, }; // 东站结算信息 // 1 [东站。100.0。东东] 00010 东东 2018-08-24 01:01 null // 2 [东站。123.0。西西] 00012 西西 2018-08-24 01:02 null /* var vehicleStationStatementsEast = new List { new VehicleStationStatement { Id = 1, Detail = "[东站。100.0。东东]", Code = "00010", Name = "东东", CreateTime = DateTime.Parse("2018-08-24 01:01:00"), UpdateTime = DateTime.Parse("2018-08-24 01:05:00") }, new VehicleStationStatement { Id = 2, Detail = "[东站。123.0。西西]", Code = "00012", Name = "西西", CreateTime = DateTime.Parse("2018-08-24 01:02:00"), UpdateTime = DateTime.Parse("2018-08-24 01:05:00") } }; */ // 西站结算信息 // 1 [西站。101.0。广州东东信息技术有限公司] 00001 广州东东信息技术有限公司 2018-08-24 01:04 null // 2 [西站。124.0。广州西西信息技术有限公司] 00002 广州西西信息技术有限公司 2018-08-24 01:03 null /* var vehicleStationStatementsWest = new List { new VehicleStationStatement { Id = 1, Detail = "[西站。101.0。广州东东信息技术有限公司]", Code = "00001", Name = "广州东东信息技术有限公司", CreateTime = DateTime.Parse("2018-08-24 01:04:00"), UpdateTime = DateTime.Parse("2018-08-24 01:05:00") }, new VehicleStationStatement { Id = 2, Detail = "[西站。124.0。广州西西信息技术有限公司]", Code = "00002", Name = "广州西西信息技术有限公司", CreateTime = DateTime.Parse("2018-08-24 01:03:00"), UpdateTime = DateTime.Parse("2018-08-24 01:05:00") } }; */ /* Parallel.Invoke( () => { // 1 [东站。99。东东] 00010 东东 2018-08-24 01:01:00 2018-08-24 01:01:00 var code = "00010"; var name = "东东"; for (var i = 0; i < 10000; i++) { using (var connection = new SqlConnection(vehicleStations[0].Token)) { var command = new SqlCommand(@" INSERT INTO [VehicleStationStatement](Detail,Code,Name,CreateTime,UpdateTime) VALUES(@Detail,@Code,@Name,@CreateTime,@UpdateTime) ", connection); command.Parameters.AddWithValue("Detail", $"[{vehicleStations[0].Name}。{new Random().Next(1, 100)}。{name}。{i}]"); command.Parameters.AddWithValue("Code", code); command.Parameters.AddWithValue("Name", name); command.Parameters.AddWithValue("CreateTime", DateTime.Now); command.Parameters.AddWithValue("UpdateTime", DateTime.Now); connection.Open(); command.ExecuteNonQuery(); } } }, () => { // 2 [东站。99。西西] 00012 西西 2018-08-24 01:01:00 2018-08-24 01:01:00 var code = "00012"; var name = "西西"; for (var i = 0; i < 20000; i++) { using (var connection = new SqlConnection(vehicleStations[0].Token)) { var command = new SqlCommand(@" INSERT INTO [VehicleStationStatement](Detail,Code,Name,CreateTime,UpdateTime) VALUES(@Detail,@Code,@Name,@CreateTime,@UpdateTime) ", connection); command.Parameters.AddWithValue("Detail", $"[{vehicleStations[0].Name}。{new Random().Next(1, 100)}。{name}。{i}]"); command.Parameters.AddWithValue("Code", code); command.Parameters.AddWithValue("Name", name); command.Parameters.AddWithValue("CreateTime", DateTime.Now); command.Parameters.AddWithValue("UpdateTime", DateTime.Now); connection.Open(); command.ExecuteNonQuery(); } } }, () => { // 1 [西站。99。广州东东信息技术有限公司] 00001 广州东东信息技术有限公司 2018-08-24 01:01:00 2018-08-24 01:01:00 var code = "00001"; var name = "广州东东信息技术有限公司"; for (var i = 0; i < 30000; i++) { using (var connection = new SqlConnection(vehicleStations[1].Token)) { var command = new SqlCommand(@" INSERT INTO [VehicleStationStatement](Detail,Code,Name,CreateTime,UpdateTime) VALUES(@Detail,@Code,@Name,@CreateTime,@UpdateTime) ", connection); command.Parameters.AddWithValue("Detail", $"[{vehicleStations[1].Name}。{new Random().Next(1, 100)}。{name}。{i}]"); command.Parameters.AddWithValue("Code", code); command.Parameters.AddWithValue("Name", name); command.Parameters.AddWithValue("CreateTime", DateTime.Now); command.Parameters.AddWithValue("UpdateTime", DateTime.Now); connection.Open(); command.ExecuteNonQuery(); } } }, () => { // 2 [西站。99。广州西西信息技术有限公司] 00002 广州西西信息技术有限公司 2018-08-24 01:01:00 2018-08-24 01:01:00 var code = "00002"; var name = "广州西西信息技术有限公司"; for (var i = 0; i < 40000; i++) { using (var connection = new SqlConnection(vehicleStations[1].Token)) { var command = new SqlCommand(@" INSERT INTO [VehicleStationStatement](Detail,Code,Name,CreateTime,UpdateTime) VALUES(@Detail,@Code,@Name,@CreateTime,@UpdateTime) ", connection); command.Parameters.AddWithValue("Detail", $"[{vehicleStations[1].Name}。{new Random().Next(1, 100)}。{name}。{i}]"); command.Parameters.AddWithValue("Code", code); command.Parameters.AddWithValue("Name", name); command.Parameters.AddWithValue("CreateTime", DateTime.Now); command.Parameters.AddWithValue("UpdateTime", DateTime.Now); connection.Open(); command.ExecuteNonQuery(); } } } ); */ var database = RedisHelper.GetConnectionMultiplexer().GetDatabase(); Parallel.ForEach(vehicleStations, (vehicleStation) => { var connection = new SqlConnection(vehicleStation.Token); var command = new SqlCommand(@" SELECT Id,Detail,Code,Name,CreateTime,UpdateTime FROM [VehicleStationStatement]", connection); connection.Open(); using (var reader = command.ExecuteReader(CommandBehavior.CloseConnection)) { while (reader.Read()) { var id = reader.GetInt32(0); var detail = reader.GetString(1); var code = reader.GetString(2); var name = reader.GetString(3); var createTime = reader.GetDateTime(4); var updateTime = reader.GetDateTime(5); var config = correctionConfigurations.SingleOrDefault(v => v.VehicleStationCode == vehicleStation.Code && v.CorrectionSettlementCompanyCode == code); if (config == null) break; // vehicleStationStatementsCenter.Enqueue(new VehicleStationStatement // { // Id = 0, // Detail = detail, // Code = config.SettlementCompanyCode, // Name = config.SettlementCompanyName, // CreateTime = createTime, // UpdateTime = updateTime // }); database.ListRightPush("VehicleStationStatement", JsonConvert.SerializeObject(new VehicleStationStatement { Id = 0, Detail = detail, Code = config.SettlementCompanyCode, Name = config.SettlementCompanyName, CreateTime = createTime, UpdateTime = updateTime })); } } }); // Console.WriteLine(vehicleStationStatementsCenter.Count); Console.WriteLine(database.ListLength("VehicleStationStatement")); Parallel.Invoke( () => { while (database.ListLength("VehicleStationStatement") > 0) { Console.WriteLine($"[01] {database.ListLeftPop("VehicleStationStatement")}"); } }, () => { while (database.ListLength("VehicleStationStatement") > 0) { Console.WriteLine($"[02] {database.ListLeftPop("VehicleStationStatement")}"); } }, () => { while (database.ListLength("VehicleStationStatement") > 0) { Console.WriteLine($"[03] {database.ListLeftPop("VehicleStationStatement")}"); } } ); } } public class SettlementCompany { public int Id { get; set; } public string Code { get; set; } public string Name { get; set; } } public class VehicleStation { public int Id { get; set; } public string Code { get; set; } public string Name { get; set; } public string Token { get; set; } } public class CorrectionConfiguration { public int Id { get; set; } public string VehicleStationCode { get; set; } public string VehicleStationName { get; set; } public string SettlementCompanyCode { get; set; } public string SettlementCompanyName { get; set; } public string CorrectionSettlementCompanyCode { get; set; } public string CorrectionSettlementCompanyName { get; set; } } public class VehicleStationStatement { public int Id { get; set; } public string Detail { get; set; } public string Code { get; set; } public string Name { get; set; } public DateTime CreateTime { get; set; } public DateTime? UpdateTime { get; set; } } public static class RedisHelper { private static readonly ConcurrentDictionary _connectionMultiplexers = new ConcurrentDictionary (); private static readonly object _locker = new object(); private static ConnectionMultiplexer _connectionMultiplexer; public static ConnectionMultiplexer GetConnectionMultiplexer(string connectionString = "localhost") { if (_connectionMultiplexer == null) { lock (_locker) { if (_connectionMultiplexer == null || _connectionMultiplexer.IsConnected == false) { if (_connectionMultiplexers.ContainsKey(connectionString) == false) { var connectionMultiplexer = ConnectionMultiplexer.Connect(connectionString); connectionMultiplexer.ConfigurationChanged += ConfigurationChanged; connectionMultiplexer.ConfigurationChangedBroadcast += ConfigurationChangedBroadcast; connectionMultiplexer.ConnectionFailed += ConnectionFailed; connectionMultiplexer.ConnectionRestored += ConnectionRestored; connectionMultiplexer.ErrorMessage += ErrorMessage; connectionMultiplexer.HashSlotMoved += HashSlotMoved; connectionMultiplexer.InternalError += InternalError; _connectionMultiplexers[connectionString] = connectionMultiplexer; } _connectionMultiplexer = _connectionMultiplexers[connectionString]; } } } return _connectionMultiplexer; } private static void ConfigurationChanged(object sender, EndPointEventArgs e) { Console.WriteLine($"ConfigurationChanged"); // e.EndPoint } private static void ConfigurationChangedBroadcast(object sender, EndPointEventArgs e) { Console.WriteLine($"ConfigurationChangedBroadcast"); } private static void ConnectionFailed(object sender, ConnectionFailedEventArgs e) { Console.WriteLine($"ConnectionFailed"); } private static void ConnectionRestored(object sender, ConnectionFailedEventArgs e) { Console.WriteLine($"ConnectionRestored"); } private static void ErrorMessage(object sender, RedisErrorEventArgs e) { Console.WriteLine($"ErrorMessage"); } private static void HashSlotMoved(object sender, HashSlotMovedEventArgs e) { Console.WriteLine($"HashSlotMoved"); } private static void InternalError(object sender, InternalErrorEventArgs e) { Console.WriteLine($"InternalError"); } }}// Install-Package StackExchange.Redis// Install-Package Newtonsoft.Json