-
Notifications
You must be signed in to change notification settings - Fork 0
/
NpgsqlConnectionPool.cs
153 lines (119 loc) · 4.23 KB
/
NpgsqlConnectionPool.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
using Npgsql;
using SafeObjectPool;
using System;
using System.Collections.Generic;
using System.Data;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading.Tasks;
namespace Npgsql {
public class NpgsqlConnectionPool : ObjectPool<NpgsqlConnection> {
internal Action availableHandler;
internal Action unavailableHandler;
public NpgsqlConnectionPool(string name, string connectionString, Action availableHandler, Action unavailableHandler) : base(null) {
var policy = new NpgsqlConnectionPoolPolicy {
_pool = this,
Name = name
};
this.Policy = policy;
policy.ConnectionString = connectionString;
this.availableHandler = availableHandler;
this.unavailableHandler = unavailableHandler;
}
public void Return(Object<NpgsqlConnection> obj, Exception exception, bool isRecreate = false) {
if (exception != null && exception is NpgsqlException) {
if (exception is System.IO.IOException) {
base.SetUnavailable(exception);
} else if (obj.Value.Ping() == false) {
base.SetUnavailable(exception);
}
}
base.Return(obj, isRecreate);
}
}
public class NpgsqlConnectionPoolPolicy : IPolicy<NpgsqlConnection> {
internal NpgsqlConnectionPool _pool;
public string Name { get; set; } = "PostgreSQL NpgsqlConnection 对象池";
public int PoolSize { get; set; } = 100;
public TimeSpan SyncGetTimeout { get; set; } = TimeSpan.FromSeconds(10);
public TimeSpan IdleTimeout { get; set; } = TimeSpan.Zero;
public int AsyncGetCapacity { get; set; } = 10000;
public bool IsThrowGetTimeoutException { get; set; } = true;
public int CheckAvailableInterval { get; set; } = 5;
private string _connectionString;
public string ConnectionString {
get => _connectionString;
set {
_connectionString = value ?? "";
Match m = Regex.Match(_connectionString, @"Maximum\s*pool\s*size\s*=\s*(\d+)", RegexOptions.IgnoreCase);
if (m.Success == false || int.TryParse(m.Groups[1].Value, out var poolsize) == false || poolsize <= 0) poolsize = 100;
PoolSize = poolsize;
var initConns = new Object<NpgsqlConnection>[poolsize];
for (var a = 0; a < poolsize; a++) try { initConns[a] = _pool.Get(); } catch { }
foreach (var conn in initConns) _pool.Return(conn);
}
}
public bool OnCheckAvailable(Object<NpgsqlConnection> obj) {
if (obj.Value.State == ConnectionState.Closed) obj.Value.Open();
var cmd = obj.Value.CreateCommand();
cmd.CommandText = "select 1";
cmd.ExecuteNonQuery();
return true;
}
public NpgsqlConnection OnCreate() {
var conn = new NpgsqlConnection(_connectionString);
return conn;
}
public void OnDestroy(NpgsqlConnection obj) {
if (obj.State != ConnectionState.Closed) obj.Close();
obj.Dispose();
}
public void OnGet(Object<NpgsqlConnection> obj) {
if (_pool.IsAvailable) {
if (obj.Value.State != ConnectionState.Open || DateTime.Now.Subtract(obj.LastReturnTime).TotalSeconds > 60 && obj.Value.Ping() == false) {
try {
obj.Value.Open();
} catch (Exception ex) {
if (_pool.SetUnavailable(ex) == true)
throw new Exception($"【{this.Name}】状态不可用,等待后台检查程序恢复方可使用。{ex.Message}");
}
}
}
}
async public Task OnGetAsync(Object<NpgsqlConnection> obj) {
if (_pool.IsAvailable) {
if (obj.Value.State != ConnectionState.Open || DateTime.Now.Subtract(obj.LastReturnTime).TotalSeconds > 60 && obj.Value.Ping() == false) {
try {
await obj.Value.OpenAsync();
} catch (Exception ex) {
if (_pool.SetUnavailable(ex) == true)
throw new Exception($"【{this.Name}】状态不可用,等待后台检查程序恢复方可使用。{ex.Message}");
}
}
}
}
public void OnGetTimeout() {
}
public void OnReturn(Object<NpgsqlConnection> obj) {
}
public void OnAvailable() {
_pool.availableHandler?.Invoke();
}
public void OnUnavailable() {
_pool.unavailableHandler?.Invoke();
}
}
public static class NpgsqlConnectionExtensions {
public static bool Ping(this NpgsqlConnection that) {
try {
var cmd = that.CreateCommand();
cmd.CommandText = "select 1";
cmd.ExecuteNonQuery();
return true;
} catch {
try { that.Close(); } catch { }
return false;
}
}
}
}