我正在练习将MMF实现为IPC，我已经通过wpf作为侦听器进行了尝试，并设置了触发的事件处理程序。
现在，我已经将相同的资源库代码移动到 Wpf 应用程序(在此之前的资源库是控制台应用程序(中，我无法触发事件处理程序。
这是我成功实现的代码，我不确定这个实现是否正确，...我唯一确定的是它发送数据，接收回复(实际上来自 WPF(并触发其事件(在控制台上(。
MMF类
public class MMFDepositT //: DiposiT<MyData>
{
private System.Threading.SendOrPostCallback callback;
public event EventHandler<MemoryMappedDataReceivedEventArgs> DataReceived;
private System.ComponentModel.AsyncOperation operation;
public virtual string DipositChlName { get; set; }
public virtual string DipositThrdName { get; set; }
public virtual int DipositSize { get; set; }
bool started;
private bool disposed;
public int ReadPosition { get; set; }
public List<string> statusSet;
private int writePosition;
public int WritePosition
{
get { return writePosition; }
set
{
if (value != writePosition)
{
this.writePosition = value;
this.accessor.Write(WritePosition + READ_CONFIRM_OFFSET, true);
}
}
}
private const int DATA_AVAILABLE_OFFSET = 0;
private const int READ_CONFIRM_OFFSET = DATA_AVAILABLE_OFFSET + 1;
private const int DATA_LENGTH_OFFSET = READ_CONFIRM_OFFSET + 1;
private const int DATA_OFFSET = DATA_LENGTH_OFFSET + 10;
public MMFDepositT(string ctrIpcChannelNameStr, string ctrIpcThreadName, int ctrMMFSize)
{
this.DipositChlName = ctrIpcChannelNameStr;
this.DipositSize = ctrMMFSize;
this.DipositThrdName = ctrIpcThreadName;
this.statusSet = new List<string>();
try
{
mmf = MemoryMappedFile.CreateOrOpen(DipositChlName, DipositSize);
accessor = mmf.CreateViewAccessor(0, DipositSize, System.IO.MemoryMappedFiles.MemoryMappedFileAccess.ReadWrite);//if (started)
//smLock = new System.Threading.Mutex(true, IpcMutxName, out locked);
ReadPosition = -1;
writePosition = -1;
this.dataToSend = new List<byte[]>();
this.callback = new System.Threading.SendOrPostCallback(OnDataReceivedInternal);
this.operation = System.ComponentModel.AsyncOperationManager.CreateOperation(null);
this.statusSet.Add("accessorOk & mmfOk");
}
catch (Exception e)
{
this.statusSet.Add("accessor Excep:-> " + e.Message);
}
// return;
//System.Threading.Thread t = new System.Threading.Thread(ReaderThread);
//t.IsBackground = true;
//t.Start();
//this.started = true;
}
public void StartReader()
{
////FieldInfo field = typeof(DipositValClF.ValueStrNameMax10SrV).GetField("Val");
////object[] attributes = field.GetCustomAttributes(typeof(MarshalAsAttribute), false);
////MarshalAsAttribute marshal = (MarshalAsAttribute)attributes[0];
// Create named MMF
if (started) return;
if (ReadPosition < 0 || writePosition < 0)
throw new ArgumentException();
System.Threading.Thread t = new System.Threading.Thread(ReaderThread);
t.IsBackground = true;
t.Start();
this.started = true;
}
private void ReaderThread(object stateInfo)
{
while (started)
{
// Checks if there is something to read.
var dataAvailable = accessor.ReadBoolean(ReadPosition + DATA_AVAILABLE_OFFSET);
if (dataAvailable)
{
// Checks how many bytes to read.
int availableBytes = accessor.ReadInt32(ReadPosition + DATA_LENGTH_OFFSET);
var bytes = new byte[availableBytes];
// Reads the byte array.
int read = accessor.ReadArray<byte>(ReadPosition + DATA_OFFSET, bytes, 0, availableBytes);
// Sets the flag used to signal that there aren't available data anymore.
accessor.Write(ReadPosition + DATA_AVAILABLE_OFFSET, false);
// Sets the flag used to signal that data has been read.
accessor.Write(ReadPosition + READ_CONFIRM_OFFSET, true);
MemoryMappedDataReceivedEventArgs args = new MemoryMappedDataReceivedEventArgs(bytes, read);
operation.Post(callback, args);
}
System.Threading.Thread.Sleep(500);
}
}
public void CloseReader()
{
started = false;
}
private System.Threading.Thread writerThread;
private List<byte[]> dataToSend;
private bool writerThreadRunning;
public void Write(byte[] data)
{
if (ReadPosition < 0 || writePosition < 0)
throw new ArgumentException();
this.statusSet.Add("ReadWrite:-> " + ReadPosition + "-" + writePosition);
// var s = (FsMomitorIPCCrier)data;
lock (this.dataToSend)
this.dataToSend.Add(data);
if (!writerThreadRunning)
{
writerThreadRunning = true;
writerThread = new System.Threading.Thread(WriterThread);
writerThread.IsBackground = true;
writerThread.Name = this.DipositThrdName;
writerThread.Start();
}
}
public void WriterThread(object stateInfo)
{
while (dataToSend.Count > 0 && !this.disposed)
{
byte[] data = null;
lock (dataToSend)
{
data = dataToSend[0];
dataToSend.RemoveAt(0);
}
while (!this.accessor.ReadBoolean(WritePosition + READ_CONFIRM_OFFSET))
System.Threading.Thread.Sleep(500);
// Sets length and write data.
this.accessor.Write(writePosition + DATA_LENGTH_OFFSET, data.Length);
this.accessor.WriteArray<byte>(writePosition + DATA_OFFSET, data, 0, data.Length);
// Resets the flag used to signal that data has been read.
this.accessor.Write(writePosition + READ_CONFIRM_OFFSET, false);
// Sets the flag used to signal that there are data avaibla.
this.accessor.Write(writePosition + DATA_AVAILABLE_OFFSET, true);
}
writerThreadRunning = false;
}
private void OnDataReceivedInternal(object state)
{
OnDataReceived(state as MemoryMappedDataReceivedEventArgs);
}
protected virtual void OnDataReceived(MemoryMappedDataReceivedEventArgs e)
{
if (e != null && DataReceived != null)
DataReceived(this, e);
}
public virtual void Close()
{
//accessor.Dispose();
//mmf.Dispose();
//smLock.Close();
started = false;
if (accessor != null)
{
try
{
accessor.Dispose();
accessor = null;
}
catch { }
}
if (this.mmf != null)
{
try
{
mmf.Dispose();
mmf = null;
}
catch { }
}
disposed = true;
GC.SuppressFinalize(this);
}
private System.IO.MemoryMappedFiles.MemoryMappedFile mmf;
private System.IO.MemoryMappedFiles.MemoryMappedViewAccessor accessor;
}
设置几个EventWaitHandle，我已经全部替换了：
private System.Threading.SendOrPostCallback callback;
public event EventHandler<MemoryMappedDataReceivedEventArgs> DataReceived;
private System.ComponentModel.AsyncOperation operation;
而且是同伴...
static bool MMfGetterTest1()
{
IpcAccessorSetting parCurSrv = new IpcAccessorSetting(IPChannelS.Debugger, IpcAccessorThreadNameS.DebuggerThrd, 0, 5000000);
MMFinterComT DebuggerInterReciver=null;
DebuggerInterReciver = new MMFinterComT(parCurSrv.Channel.ToString(), parCurSrv.AccThreadName.ToString(), 10000000);
DebuggerInterReciver.ReadPosition = parCurSrv.AccessorSectorsSets.DepoSects.Getter.Read;
DebuggerInterReciver.WritePosition = parCurSrv.AccessorSectorsSets.DepoSects.Getter.Write;
DebuggerInterReciver.flagReciver1 = new EventWaitHandle(false, EventResetMode.ManualReset, DebuggerInterReciver.DipositThrdName);
DebuggerInterReciver.flagReciver2 = new EventWaitHandle(false, EventResetMode.AutoReset, IpcAccessorThreadNameS.DebuggerThrdCurProj.ToString());
DebuggerInterReciver.flagCaller1 = new EventWaitHandle(false, EventResetMode.ManualReset, IpcAccessorThreadNameS.DebuggerThrdCurProj.ToString());
DebuggerInterReciver.flagCaller2 = new EventWaitHandle(false, EventResetMode.ManualReset, IpcAccessorThreadNameS.DebuggerThrdCurProj.ToString());
System.Windows.Application.Current.Dispatcher.Invoke(new Action(() =>
MbxTwDbSrv.Show(320, "Debugger Reciver Online ", "V", "Debugger Reciver Ready onrnDebuggerInterReciver.ReadPosition " + DebuggerInterReciver.ReadPosition + "rnDebuggerGetterepo.WritePosition " + DebuggerInterReciver.WritePosition)), System.Windows.Threading.DispatcherPriority.Normal);
DebuggerInterReciver.flagCaller1.Set();
DebuggerInterReciver.flagReciver1.WaitOne();
DebuggerInterReciver.flagReciver1.Close();
DebuggerInterReciver.flagReciver2.WaitOne(200);
bool exit = false; int trys = 0;
while(!exit)
{
DebuggerInterReciver.StartReader();
System.Windows.Application.Current.Dispatcher.Invoke(new Action(() =>
MbxTwDbSrv.Show(336, "Debugger Reciver Reading Failure ", "X", "Debugger Reciver read status on try: # "+ ++trys + " "+DebuggerInterReciver.IntercomStatus)), System.Windows.Threading.DispatcherPriority.Normal);
DebuggerInterReciver.flagReciver2.WaitOne(700);
exit = DebuggerInterReciver.IntercomStatus==RobCs511C.AppMods.Selectors.IPC.IpcMMFinterComSF.MMFinterComTStatus.FinishedReading;
}
//DebuggerInterReciver.flagCaller.Set();
DebuggerInterReciver.flagReciver2.Close();
DebuggerInterReciver.flagCaller2.Set();
var x = DebuggerInterReciver.ReadData.BytesToIpcCarier().AsDebggerCarier();
System.Windows.Application.Current.Dispatcher.Invoke(new Action(() =>
MbxTwDbSrv.Show(337, "Debugger Reciver OnReciveMessage: ", "V", "Debugger Reciver Has Recived Data: " + x.DvalStr +", Len= "+ x.DvalIntArr.Length)), System.Windows.Threading.DispatcherPriority.Normal);
DebuggerInterReciver.reading = false;
return true;
}
我的新MMF可以轻松获取10G的对象，我在眨眼间做了5G，全双工进程间通信在这里完成。