我正在开发一个 .net core web 应用程序,希望能够监听 PostgreSQL 数据库并获取表格上的任何更改。
根据我的研究,我需要使用 SignalR Core。我尝试了一些 SignalR 的示例应用程序,比如聊天应用程序,但它们都没有监听数据库。我找不到这方面的任何示例。
-必须在 PostgreSQL 数据库上设置触发器吗?
-监听器必须在代码端吗?
-我该如何使用 SignalR Core?
请给我提供一种解决方法。非常感谢。
我正在开发一个 .net core web 应用程序,希望能够监听 PostgreSQL 数据库并获取表格上的任何更改。
根据我的研究,我需要使用 SignalR Core。我尝试了一些 SignalR 的示例应用程序,比如聊天应用程序,但它们都没有监听数据库。我找不到这方面的任何示例。
-必须在 PostgreSQL 数据库上设置触发器吗?
-监听器必须在代码端吗?
-我该如何使用 SignalR Core?
请给我提供一种解决方法。非常感谢。
这个示例适用于 asp.net core 3.0+。完整的代码如下。
步骤1. 在PostgreSql上创建一个触发器以便监听操作。
create trigger any_after_alarm_speed after
insert
or
delete
or
update
on
public.alarm_speed for each row execute procedure alarm_speedf();
第二步。在Postgresql上创建存储过程。
CREATE OR REPLACE FUNCTION public.alarm_speedf()
RETURNS trigger
LANGUAGE plpgsql
AS $function$
BEGIN
IF TG_OP = 'INSERT' then
PERFORM pg_notify('notifyalarmspeed', format('INSERT %s %s', NEW.alarm_speed_id,
NEW.alarm_speed_date));
ELSIF TG_OP = 'UPDATE' then
PERFORM pg_notify('notifyalarmspeed', format('UPDATE %s %s', OLD.alarm_speed_id,
OLD.alarm_speed_date));
ELSIF TG_OP = 'DELETE' then
PERFORM pg_notify('notifyalarmspeed', format('DELETE %s %s', OLD.alarm_speed_id,
OLD.alarm_speed_date));
END IF;
RETURN NULL;
END;
$function$;
步骤 3. 创建中心
public class speedalarmhub : Hub
{
private IMemoryCache _cache;
`private IHubContext<speedalarmhub> _hubContext;
public speedalarmhub(IMemoryCache cache, IHubContext<speedalarmhub> hubContext)
{
_cache = cache;
_hubContext = hubContext;
}
public async Task SendMessage()
{
if (!_cache.TryGetValue("SpeedAlarm", out string response))
{
SpeedListener speedlist = new SpeedListener(_hubContext,_cache);
speedlist.ListenForAlarmNotifications();
string jsonspeedalarm = speedlist.GetAlarmList();
_cache.Set("SpeedAlarm", jsonspeedalarm);
await Clients.All.SendAsync("ReceiveMessage", _cache.Get("SpeedAlarm").ToString());
}
else
{
await Clients.All.SendAsync("ReceiveMessage", _cache.Get("SpeedAlarm").ToString());
}
}
}
步骤4. 创建监听器控制器
public class SpeedListener :Controller
{
private IHubContext<speedalarmhub> _hubContext;
private IMemoryCache _cache;
public SpeedListener(IHubContext<speedalarmhub> hubContext,IMemoryCache cache)
{
_hubContext = hubContext;
_cache = cache;
}
static string GetConnectionString()
{
var csb = new NpgsqlConnectionStringBuilder
{
Host = "yourip",
Database = "yourdatabase",
Username = "yourusername",
Password = "yourpassword",
Port = 5432,
KeepAlive = 30
};
return csb.ConnectionString;
}
public void ListenForAlarmNotifications()
{
NpgsqlConnection conn = new NpgsqlConnection(GetConnectionString());
conn.StateChange += conn_StateChange;
conn.Open();
var listenCommand = conn.CreateCommand();
listenCommand.CommandText = $"listen notifyalarmspeed;";
listenCommand.ExecuteNonQuery();
conn.Notification += PostgresNotificationReceived;
_hubContext.Clients.All.SendAsync(this.GetAlarmList());
while (true)
{
conn.Wait();
}
}
private void PostgresNotificationReceived(object sender, NpgsqlNotificationEventArgs e)
{
string actionName = e.Payload.ToString();
string actionType = "";
if (actionName.Contains("DELETE"))
{
actionType = "Delete";
}
if (actionName.Contains("UPDATE"))
{
actionType = "Update";
}
if (actionName.Contains("INSERT"))
{
actionType = "Insert";
}
_hubContext.Clients.All.SendAsync("ReceiveMessage", this.GetAlarmList());
}
public string GetAlarmList()
{
var AlarmList = new List<AlarmSpeedViewModel>();
using (NpgsqlCommand sqlCmd = new NpgsqlCommand())
{
sqlCmd.CommandType = CommandType.StoredProcedure;
sqlCmd.CommandText = "sp_alarm_speed_process_get";
NpgsqlConnection conn = new NpgsqlConnection(GetConnectionString());
conn.Open();
sqlCmd.Connection = conn;
using (NpgsqlDataReader reader = sqlCmd.ExecuteReader())
{
while (reader.Read())
{
AlarmSpeedViewModel model = new AlarmSpeedViewModel();
model.alarm_speed_id = reader.GetInt32(0);
// you must fill your model items
AlarmList.Add(model);
}
reader.Close();
conn.Close();
}
}
_cache.Set("SpeedAlarm", SerializeObjectToJson(AlarmList));
return _cache.Get("SpeedAlarm").ToString();
}
public String SerializeObjectToJson(Object alarmspeed)
{
try
{
var jss = new JavaScriptSerializer();
return jss.Serialize(alarmspeed);
}
catch (Exception) { return null; }
}
private void conn_StateChange(object sender, System.Data.StateChangeEventArgs e)
{
_hubContext.Clients.All.SendAsync("Current State: " + e.CurrentState.ToString() + " Original State: " + e.OriginalState.ToString(), "connection state changed");
}
}
第五步 调用中心节点。<script src="~/lib/signalr.js"></script>
<script type="text/javascript">
// Start the connection.
var connection = new signalR.HubConnectionBuilder()
.withUrl('/speedalarmhub')
.build();
connection.on('ReceiveMessage', function (message) {
var encodedMsg = message;
// Add the message to the page.
});
// Transport fallback functionality is now built into start.
connection.start()
.then(function () {
console.log('connection started');
connection.invoke('SendMessage');
})
.catch(error => {
console.error(error.message);
});
步骤6。在启动时添加以下代码配置服务
public void ConfigureServices(IServiceCollection services)
{
services.AddControllersWithViews();
services.AddSignalR();
services.AddMemoryCache();
}
步骤7:在配置方法中添加以下代码
app.UseEndpoints(endpoints =>
{
endpoints.MapControllerRoute(
name: "default",
pattern: "{controller=Home}/{action=Index}/{id?}");
endpoints.MapHub<speedalarmhub>("/speedalarmhub");
});
pg_notify(text, text)
发送通知,如下所示。
函数
CREATE OR REPLACE FUNCTION mytestfunc() RETURNS TRIGGER AS $$
BEGIN
IF TG_OP = 'INSERT' then
PERFORM pg_notify('notifytesttable', 'new record inserted');
ELSIF TG_OP = 'UPDATE' then
PERFORM pg_notify('notifytesttable', 'updated');
ELSIF TG_OP = 'DELETE' then
PERFORM pg_notify('notifytesttable', 'deleted');
END IF;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
触发器
CREATE TRIGGER any_after_testtable
AFTER INSERT OR DELETE OR UPDATE
ON testtable
FOR EACH ROW
EXECUTE PROCEDURE mytestfunc();
在您的客户端应用程序代码中,您可以监听和接收来自PostgreSQL的通知。
conn.Open();
conn.Notification += Conn_Notification;
using (var cmd = new NpgsqlCommand("LISTEN notifytesttable", conn))
{
cmd.ExecuteNonQuery();
}
Notification
事件处理程序中,您可以调用 SignalR hub 方法向 SignalR 客户端推送通知。private static void Conn_Notification(object sender, NpgsqlNotificationEventArgs e)
{
var notification_payload = e.Payload;
//code logic here
//call hub method to push PostgreSQL notifications that you received to SignalR client users
}
测试结果
若要了解 PostgreSQL 的 LISTEN 和 NOTIFY 功能的详细信息,请参阅以下链接。
https://www.postgresql.org/docs/current/sql-notify.html
https://www.npgsql.org/doc/wait.html#processing-of-notifications
changeOccured
的表,并每隔几秒钟检查该表即可),如果发生了变化,则使用 SignalR 通知客户端。 - Veljko89