Examples of channel communication:
using ConsoleApp2;
using ;
using ;
var queue = new BlockingCollection<Message>(new ConcurrentQueue<Message>());
var opt = new BoundedChannelOptions(10)
{
FullMode = ,
SingleReader = true,
SingleWriter = true,
Capacity = 100 //Maximum capacity};
// Limitedvar channelTest = <Message>(opt);
//Unlimitedvar channel = <Message>();
var sender1 = SendMessageThreadAsync(, 1);
var sender2 = SendMessageThreadAsync(, 2);
var receiver1 = ReceiveMessageThreadAsync(, 3);
var receiver2 = ReceiveMessageThreadAsync(, 4);
//await sender;
// make sure all messages are received
await (sender1, sender2);
();
await (receiver1, receiver2);
//await receiver;
("Press any key to exit...");
();
async Task SendMessageThreadAsync(ChannelWriter<Message> writer, int id)
{
for (int i = 0; i < 20; i++)
{
await (new Message(id, ()));
($"Thread {id} sent {i}");
await (100);
}
}
async Task ReceiveMessageThreadAsync(ChannelReader<Message> reader, int id)
{
//try
//{
// while (!)
// {
// var message = await ();
// ($"Thread {id} received {}");
// }
//}
//catch (Exception ex)
//{
// ($"Thread {id} channel closed:{}");
//}
await foreach (var message in ())
{
($"Thread {id} received {}");
}
}
record Message(int FromId, string Content);
Transformed into Plc example
record PlcDataMessage
{
public bool IsConnected { get; init; }
public DbData DbData { get; init; }
// You can add other information that needs to be passed}
// Create a boundless channel to send and receive messagesvar plcDataChannel = <PlcDataMessage>();
// Start a new task to simulate PLC data reading(async () =>
{
var cts = new CancellationTokenSource(); // Assume that you already have a cancel token source while (!)
{
try
{
// ... The code connecting the PLC is omitted, and this part of the logic remains unchanged...
if (MyIsConnected)
{
DbData dbDataTemp = await <DbData>(42, 0);
// Heartbeat and other operations...
// Construct the message and send it to Channel var message = new PlcDataMessage
{
IsConnected = MyIsConnected,
DbData = dbDataTemp
};
await (message, );
}
// ... Other logic remains the same ... }
catch (Exception ex)
{
// Handle the exception and reconnect the PLC (if required) // ...
// A special message can be sent through the Channel to indicate that the connection has been disconnected or an error has occurred // This part of the logic is omitted here
// Try again after sleeping for a while await (2000, );
}
}
// Notify Channel after completion no more data will be sent ();
}, , , );
// Read data in Channel in another task or thread(async () =>
{
await foreach (var message in ())
{
if ()
{
lock (lockObj)
{
// Update dbData, here assuming that dbData is a thread-safe object or structure dbData.Str_S = .Str_S.Trim();
// ... Update other properties ... }
// Process the read data... }
else
{
// Handle PLC disconnection... }
}
// Reading is completed, Channel is closed ("PLC data has been read.");
}, );
// ... Other codes, such as waiting for all tasks to complete, processing cancellation logic, etc. ...using System;
using ;
using ;
using ;
// ... Other necessary references and type definitions ...
// Create a boundless channel to send and receive messagesvar plcDataChannel = <PlcDataMessage>();
// Cancel the token sourcevar cts = new CancellationTokenSource();
// Start a new task to simulate PLC data reading(async () =>
{
Plc s7Plc = null;
bool MyIsConnected = false;
int errorTimes = 0;
try
{
while (!)
{
if (s7Plc == null || !MyIsConnected)
{
// Try to connect to the PLC (omit the specific implementation) // ...
if (MyIsConnected)
{
// Connection is successful, send a successful connection message (if required) // ...
}
}
else
{
try
{
// Read PLC data (excluding specific implementations) DbData dbDataTemp = await <DbData>(42, 0, );
// Heartbeat and other operations...
// Construct the message and send it to Channel var message = new PlcDataMessage { IsConnected = MyIsConnected, DbData = dbDataTemp };
await (message, );
errorTimes = 0; // Reset the error counter }
catch (Exception ex)
{
errorTimes++;
// Handle exceptions (such as logging) // ...
// After a certain number of errors are reached, close the PLC connection and reset if (errorTimes > someThreshold)
{
s7Plc?.Close();
s7Plc = null;
MyIsConnected = false;
// You can choose to send a disconnected message to the Channel }
// Try again after sleeping for a while await (2000, );
}
}
// Some delays can be added to reduce the frequency of the cycle await (somePollingInterval, );
}
}
catch (OperationCanceledException)
{
// Cancellation is expected and no additional processing is required }
finally
{
// Make sure to close the PLC connection and Channel writer s7Plc?.Close();
();
}
}, );
// Read data in Channel in another task or thread(async () =>
{
await foreach (var message in ())
{
if ()
{
// Update dbData (here assumes that dbData is a thread-safe object or structure) // Add appropriate synchronization mechanism as needed // ...
// Process the read data... }
else
{
// Handle PLC disconnection... }
}
// Reading is completed, Channel is closed ("PLC data has been read.");
}, );
// ... Other codes, such as waiting for all tasks to complete, processing cancellation logic, etc. ...
// Cancel the task at some appropriate moment// ();
// Wait for all tasks to complete (if requiredExpand: C# Channel implements inter-thread communication
C# Channel implements inter-thread communication
Synchronous method implementation:
using System;
using ;
using ;
using ;
using ;
using ;
using ;
using ;
namespace ConsoleApp1
{
public class ChannelDemo
{
static Channel<Message> channel1 = <Message>();
public static void Main2()
{
(1);
(2);
(3);
();
(3000);
();
();
();
();
();
}
static Thread sender = new Thread(SendMsg);
static Thread receive1 = new Thread(ReceiveMsg);
static Thread receive2 = new Thread(ReceiveMsg);
static void SendMsg(object id)
{
for (int i = 0; i < 20; i++)
{
if ((new Message((int)id, ())))
{
($"【Thread{id}】Sent【{i}】");
}
}
}
static void ReceiveMsg(object id)
{
try
{
while (true)
{
if ((out Message message))
{
($"【Thread{id}】from【Thread{}】Received【{}】");
}
(1);
}
}
catch (ThreadInterruptedException ex)
{
($"End of Reception");
}
}
}
}
Asynchronous method:
using System;
using ;
using ;
using ;
using ;
using ;
using ;
using ;
using ;
namespace ConsoleApp1
{
public class ChannelDemo2
{
static Channel<Message> channel1 = <Message>();
public static async void Main2()
{
await (sender, sender2);
();
await (receive1, receive2);
();
}
static Task sender = SendMsgAsync(, 1);
static Task sender2 = SendMsgAsync(, 4);
static Task receive1 = ReceiveMsgAsync(, 2);
static Task receive2 = ReceiveMsgAsync(, 3);
static async Task SendMsgAsync(ChannelWriter<Message> writer, int id)
{
for (int i = 0; i < 20; i++)
{
await (new Message((int)id, ()));
($"【Thread{id}】Sent【{i}】");
}
}
static async Task ReceiveMsgAsync(ChannelReader<Message> reader,int id)
{
try
{
while (!)
{
Message message = await ();
($"【Thread{id}】from【Thread{}】Received【{}】");
}
}
catch (ChannelClosedException ex)
{
($"ChannelClosed Received End");
}
}
}
}
When instantiating the Channel, you can also pass an Options, where the message capacity, whether multiple senders and recipients can be defined.
The above is the detailed content of C# using channel to implement communication between Plc asynchronous tasks. For more information about C# channel Plc asynchronous communication, please pay attention to my other related articles!