环形缓冲区.ringbuff，C#和java
环形缓冲，
本质就是队列fifo，先进先出的特殊版本，环形队列，是用空间得到了顺序存储的快索引的优点，又避免了删除，移动数据的缺点。并且还享受了单生产/单消费，2线程的无锁线程优势。十分完美。
1.面对频率超级频繁的读写，环形缓冲修改为固定大小的队列，不添加操作，并且使用后，也不删除和移动。坏处是，必须预设内存空间。所以适合超级频繁的读写，反正基本是一直占用。
2.因为要实现环形，所以一般有2个哨兵，如果是2个线程，一个读一个写。环形缓冲，还可以避免线程锁。非常棒。即所谓单生产/单消费模式的共享队列是不需要加锁同步的
3.固定大小。所以数组 xxx[]，非常适合改造为环形缓冲。char[] 改为字符群缓冲 . object[] 改为对象群缓冲
花了点时间，写了下。测试用例测试了下。没发现问题。
感觉缓存区必须根据实际情况写特例。
这里缓存区，适合 读必须大于写。
特点
1.写满会 自动跳过。 所以需要根据情况，定制缓存区大小，或者改写为自动扩大缓存区。
2.读，如果无数据会直接出去，无阻塞.
3. 如果可行的话，可以使用多个ringbuff，来代替多线程的方案
需改进：
定义一个最大缓存区大小。让缓存区可以自动扩大，到最大就不扩大了。
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace BaseLib
{
/// <summary>
/// 环形缓冲区. c# byte== c++ unsign char . c# char==c ++ char.
/// 环形缓冲结构，不考虑在读或写某端有多线程的情况，因为环形缓冲结构就是为了不锁下的性能，适合读写同一个线程，或者读写各一个线程。
/// </summary>
public class RingBuff
{
private byte[] ringBuff;
private int nextWritePos;
private int nextReadPos;
private int buffSize;
private int capcity;
public RingBuff(int _size)
{
capcity=buffSize = _size;
ringBuff = new byte[buffSize];
nextWritePos = 0;
nextReadPos = 0;
}
/// <summary>
/// 写入缓存
/// </summary>
/// <param name="_buff">要写入的数据</param>
/// <returns>是否写入</returns>
public bool WriteBuff(byte[] _buff)
{
bool ret = false;
int bsize = _buff.Length;
if (capcity < bsize)
{
ret = false;
}
else
{
int rightLeft = buffSize - nextWritePos;
//need reture to head
if (bsize > rightLeft)
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, rightLeft);
Array.Copy(_buff, rightLeft, ringBuff, 0, bsize - rightLeft);
nextWritePos = bsize - rightLeft;
}
else if (bsize == rightLeft)
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
nextWritePos = 0;
}
else
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
nextWritePos += bsize;
}
capcity = capcity - bsize;
ret = true;
}
return ret;
}
/// <summary>
/// 写入缓存，指定读取源的长度.
/// </summary>
/// <param name="_buff"></param>
/// <param name="len"></param>
/// <returns></returns>
public bool WriteBuff(byte[] _buff, int len)
{
int bsize = len;// _buff.Length;
bsize = len > _buff.Length ? _buff.Length : bsize;
if (capcity < bsize)
{
return false;
}
else
{
int rightLeft = buffSize - nextWritePos;
//need reture to head
if (bsize > rightLeft)
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, rightLeft);
Array.Copy(_buff, rightLeft, ringBuff, 0, bsize - rightLeft);
nextWritePos = bsize - rightLeft;
}
else if (bsize == rightLeft)
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
nextWritePos = 0;
}
else
{
Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
nextWritePos += bsize;
}
capcity -= bsize;
return true;
}
}
/// <summary>
/// 读缓存,是否预读，而不移动读哨兵.
/// </summary>
/// <param name="readbuff">读入的临时缓存</param>
/// <returns>实际读出的数据长度</returns>
public int ReadBuff(ref byte[] readbuff, bool MovePosition = true)
{
int len = readbuff.Length;
int enableread = buffSize - capcity;
Array.Clear(readbuff, 0, len);
if (len <= 0 || enableread <= 0)
{
return 0;
}
else
{
int realRead = enableread >= len ? len : enableread;
int rightLeft = buffSize - nextReadPos;
if (realRead > rightLeft)
{
Array.Copy(ringBuff, nextReadPos, readbuff, 0, rightLeft);
Array.Copy(ringBuff, 0, readbuff, rightLeft, realRead - rightLeft);
if (MovePosition)
{
nextReadPos = realRead - rightLeft;
}
}
else if (realRead == rightLeft)
{
Array.Copy(ringBuff, nextReadPos, readbuff, 0, realRead);
if (MovePosition)
{
nextReadPos = 0;
}
}
else
{
Array.Copy(ringBuff, nextReadPos, readbuff, 0, realRead);
if (MovePosition)
{
nextReadPos += realRead;
}
}
capcity += realRead;
return realRead;
}
}
public string GetBuffInfo()
{
StringBuilder sb = new StringBuilder();
sb.Append(System.Threading.Thread.CurrentThread.Name + ":writePositon:" + nextWritePos.ToString());
sb.Append(". readPosition:" + nextReadPos.ToString());
sb.Append(". size:");
sb.Append(buffSize);
sb.Append(" .enable read:" + (buffSize-capcity).ToString() + ".enable write:" + capcity.ToString());
sb.Append(System.Environment.NewLine);
sb.Append("1-10 byte: ");
for (int i = 0; i < 10; i++)
{
sb.Append(ringBuff[i].ToString("x")+"|");
}
return sb.ToString();
}
private RingBuff()
{ }
}
}测试例子
static void TestRingBuff()
{
int buffSize = 10;
BaseLib.RingBuff rb = new BaseLib.RingBuff(buffSize);
Console.WriteLine(rb.GetBuffInfo());
////测试读满，和写满。
//for (int i = 0; i < 12; i++)
//{
// rb.WriteBuff(new byte[] { (byte)testChar });
// testChar++;
// Console.WriteLine(rb.GetBuffInfo());
//}
//byte[] tempReadBuff = new byte[1];
//for (int i = 0; i < 15; i++)
//{
// rb.ReadBuff(ref tempReadBuff);
// Console.WriteLine(tempReadBuff[0].ToString("x"));
// Console.WriteLine(rb.GetBuffInfo());
//}
//////字符是依次 + 1写入缓存。所以测试首先关注各种情况下，是否读出的数据是联系的。就可大概确定数据是没有丢失和跳跃的。
////测试读慢，写快.
//while (true)
//{
// bool ret = rb.WriteBuff(new byte[] { (byte)testChar, (byte)++testChar, (byte)++testChar });
// if (ret == false)
// {
// --testChar;
// --testChar;
// --testChar;
// }
// Console.WriteLine(rb.GetBuffInfo());
// byte[] tempReadBuff = new byte[2];
// rb.ReadBuff(ref tempReadBuff);
// Console.WriteLine(tempReadBuff[0].ToString("x") + "." + tempReadBuff[1].ToString("x"));
// ++testChar;
// System.Threading.Thread.Sleep(1000);
//}
////测试读快，慢写
//while (true)
//{
// bool ret = rb.WriteBuff(new byte[] { (byte)testChar, (byte)++testChar });
// if (ret == false)
// {
// --testChar;
// --testChar;
// }
// Console.WriteLine(rb.GetBuffInfo());
// byte[] tempReadBuff = new byte[3];
// rb.ReadBuff(ref tempReadBuff);
// Console.WriteLine(tempReadBuff[0].ToString() + "." + tempReadBuff[1].ToString() + "." + tempReadBuff[2].ToString());
// Console.WriteLine(rb.GetBuffInfo());
// ++testChar;
// System.Threading.Thread.Sleep(1000);
//}
//////随机测试。
//while (true)
//{
// bool ret = rb.WriteBuff(new byte[] { (byte)testChar, (byte)++testChar, (byte)++testChar });
// if (ret == false)
// {
// --testChar;
// --testChar;
// --testChar;
// }
// Console.WriteLine(rb.GetBuffInfo());
// Random rd = new Random();
// int rint = (rd.Next() + 2) % 10;
// byte[] tempReadBuff = new byte[rint];
// Console.Write("read " + rint.ToString("x") + " :");
// rb.ReadBuff(ref tempReadBuff);
// for (int i = 0; i < rint; ++i)
// {
// Console.Write(tempReadBuff[i].ToString("x") + " ");
// }
// Console.Write("\r\n");
// Console.WriteLine(rb.GetBuffInfo());
// ++testChar;
// System.Threading.Thread.Sleep(1000);
//}
Console.Read();
}java 版本
明显发现java和c#基本语法就像双胞胎。为什么好多人包括之前的自己，会排斥另一种语言呢?
特定：读完，写满，之后都不工作了。
unusedSize;//冗余这个字段既可以提高效率，也可以区分空还是满的情况（单靠2个指针区分不了）。
package com.datastructurn;
public class RingBuff
{
private byte[] ringBuff;
private Integer nextWritePosition;
private Integer nextReadPosition;
private Integer buffSize;
private Integer unusedSize;//冗余这个字段既可以提高效率，也可以区分空还是满的情况（单靠2个指针区分不了）。
public RingBuff(int _size)
{
buffSize= unusedSize=_size;
ringBuff=new byte[buffSize];
nextReadPosition=nextWritePosition=0;
}
public boolean WriteBuff(byte[] _buff)
{
boolean ret = false;
int bsize = _buff.length;
if (unusedSize < bsize)
{
ret = false;
}
else
{
int rightLeft = buffSize - nextWritePosition;
//need reture to head
if (bsize > rightLeft)
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, rightLeft);
System.arraycopy(_buff, rightLeft, ringBuff, 0, bsize - rightLeft);
nextWritePosition = bsize - rightLeft;
}
else if (bsize == rightLeft)
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, bsize);
nextWritePosition = 0;
}
else
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, bsize);
nextWritePosition += bsize;
}
unusedSize = unusedSize - bsize;
ret = true;
}
return ret;
}
public boolean WriteBuff(byte[] _buff, int len)
{
int bsize = len;// _buff.Length;
bsize = len > _buff.length ? _buff.length : bsize;
if (unusedSize < bsize)
{
return false;
}
else
{
int rightLeft = buffSize - nextWritePosition;
//need reture to head
if (bsize > rightLeft)
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, rightLeft);
System.arraycopy(_buff, rightLeft, ringBuff, 0, bsize - rightLeft);
nextWritePosition = bsize - rightLeft;
}
else if (bsize == rightLeft)
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, bsize);
nextWritePosition = 0;
}
else
{
System.arraycopy(_buff, 0, ringBuff, nextWritePosition, bsize);
nextWritePosition += bsize;
}
unusedSize -= bsize;
return true;
}
}
/// <summary>
/// 读缓存,是否预读，而不移动读哨兵.
/// </summary>
/// <param name="readbuff">读入的临时缓存</param>
/// <returns>实际读出的数据长度</returns>
public int ReadBuff(byte[] readbuff, Boolean MovePosition )
{
if(MovePosition==null)
{
MovePosition=true;
}
int len = readbuff.length;
int enableread = buffSize - unusedSize;
//Array.Clear(readbuff, 0, len);
if (len <= 0 || enableread <= 0)
{
return 0;
}
else
{
int realRead = enableread >= len ? len : enableread;
int rightLeft = buffSize - nextReadPosition;
if (realRead > rightLeft)
{
System.arraycopy(ringBuff, nextReadPosition, readbuff, 0, rightLeft);
System.arraycopy(ringBuff, 0, readbuff, rightLeft, realRead - rightLeft);
if (MovePosition)
{
nextReadPosition = realRead - rightLeft;
}
}
else if (realRead == rightLeft)
{
System.arraycopy(ringBuff, nextReadPosition, readbuff, 0, realRead);
if (MovePosition)
{
nextReadPosition = 0;
}
}
else
{
System.arraycopy(ringBuff, nextReadPosition, readbuff, 0, realRead);
if (MovePosition)
{
nextReadPosition += realRead;
}
}
unusedSize += realRead;
return realRead;
}
}
public String GetBuffInfo()
{
StringBuilder sb = new StringBuilder();
sb.append(":writePositon:" + nextWritePosition);
sb.append(". readPosition:" + nextReadPosition);
sb.append(". size:");
sb.append(buffSize);
sb.append(" .enable read:" + (buffSize-unusedSize) + ".enable write:" + unusedSize);
sb.append("\r\n");
sb.append("1-10 byte: ");
for (int i = 0; i < 10; i++)
{
sb.append(ringBuff[i]);
}
return sb.toString();
}
}共享内存
using System;
using System.Collections.Generic;
using System.IO;
using System.IO.MemoryMappedFiles;
using System.Linq;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading.Tasks;
namespace control
{
public class ShareMemory
{
[DllImport("kernel32.dll", EntryPoint = "CopyMemory", SetLastError = false)]
public static extern void CopyMemory(IntPtr dest, IntPtr src, uint count);
//int+int+10*10=108int size = Marshal.SizeOf(typeof(char));
public void CreateShareFile(int size)
{
using (var mmf = MemoryMappedFile.CreateFromFile(@"c:\aa.data", FileMode.Open, "test", size+12))
{
//通过指定的 偏移量和大小 创建内存映射文件视图服务器
using (var accessor = mmf.CreateViewAccessor()) //偏移量，可以控制数据存储的内存位置；大小，用来控制存储所占用的空间
{
accessor.Write(0, 16);//nextWritePos
accessor.Write(4, 16);//nextReadPos;
accessor.Write(8, size);//buffSize
accessor.Write(12, size);//capcity
}
}
}
public int GetNextWritePos(MemoryMappedViewAccessor mmv)
{
return mmv.ReadInt32(0);
}
public void SetNextWritePos(MemoryMappedViewAccessor mmv,int value)
{
mmv.Write(0, value);
}
public int GetnextReadPos(MemoryMappedViewAccessor mmv)
{
return mmv.ReadInt32(4);
}
public void SetnextReadPos(MemoryMappedViewAccessor mmv, int value)
{
mmv.Write(4, value);
}
public int GetbuffSize(MemoryMappedViewAccessor mmv)
{
return mmv.ReadInt32(8);
}
public int Getcapcity(MemoryMappedViewAccessor mmv)
{
return mmv.ReadInt32(12);
}
public void Setcapcity(MemoryMappedViewAccessor mmv, int value)
{
mmv.Write(12, value);
}
public IntPtr changepd(byte[] write)
{
IntPtr write_data = Marshal.AllocHGlobal(write.Length);
Marshal.Copy(write, 0, write_data, write.Length);
Marshal.FreeHGlobal(write_data );
return write_data;
}
public bool Write(byte[] _buff)
{
bool ret = false;
using (var mmf = MemoryMappedFile.CreateFromFile(@"c:\aa.data", FileMode.Open, "test", 108))
{
using (var accessor = mmf.CreateViewAccessor())
{
int bsize = _buff.Length;
if (Getcapcity(accessor) < bsize)
{
ret = false;
}
else
{
int rightLeft = GetbuffSize(accessor) - GetnextReadPos(accessor);
if (bsize > rightLeft)
{
unsafe
{
byte* pb = (byte*)IntPtr.Zero;
accessor.SafeMemoryMappedViewHandle.AcquirePointer(ref pb);
CopyMemory((IntPtr)(pb), changepd(_buff), (uint)_buff.Length);
}
//Array.Copy(_buff, 0, ringBuff, nextWritePos, rightLeft);
//Array.Copy(_buff, rightLeft, ringBuff, 0, bsize - rightLeft);
SetNextWritePos(accessor, bsize - rightLeft);
}
else if (bsize == rightLeft)
{
//Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
SetNextWritePos(accessor, 16);
}
else
{
//Array.Copy(_buff, 0, ringBuff, nextWritePos, bsize);
SetNextWritePos(accessor,GetnextReadPos(accessor)+bsize);
}
Setcapcity(accessor, Getcapcity(accessor) - bsize);
ret = true;
}
}
}
return ret;
}
}
}- 上一篇 »PHP缓冲区强制及时输出
- 下一篇 »php 开启缓冲,页面纯静态化