版本一：
datafile1.go：
package v1
import (
"errors"
"io"
"os"
"sync"
)
// Data 代表数据的类型。
type Data []byte
// DataFile 代表数据文件的接口类型。
type DataFile interface {
// Read 会读取一个数据块。
Read() (rsn int64, d Data, err error)
// Write 会写入一个数据块。
Write(d Data) (wsn int64, err error)
// RSN 会获取最后读取的数据块的序列号。
RSN() int64
// WSN 会获取最后写入的数据块的序列号。
WSN() int64
// DataLen 会获取数据块的长度。
DataLen() uint32
// Close 会关闭数据文件。
Close() error
}
// myDataFile 代表数据文件的实现类型。
type myDataFile struct {
f *os.File // 文件。
fmutex sync.RWMutex // 被用于文件的读写锁。
woffset int64 // 写操作需要用到的偏移量。
roffset int64 // 读操作需要用到的偏移量。
wmutex sync.Mutex // 写操作需要用到的互斥锁。
rmutex sync.Mutex // 读操作需要用到的互斥锁。
dataLen uint32 // 数据块长度。
}
// NewDataFile 会新建一个数据文件的实例。
func NewDataFile(path string, dataLen uint32) (DataFile, error) {
f, err := os.Create(path)
if err != nil {
return nil, err
}
if dataLen == 0 {
return nil, errors.New("Invalid data length!")
}
df := &myDataFile{f: f, dataLen: dataLen}
return df, nil
}
func (df *myDataFile) Read() (rsn int64, d Data, err error) {
// 读取并更新读偏移量。
var offset int64
df.rmutex.Lock()
offset = df.roffset
df.roffset += int64(df.dataLen)
df.rmutex.Unlock()
//读取一个数据块。
rsn = offset / int64(df.dataLen)
bytes := make([]byte, df.dataLen)
for {
df.fmutex.RLock()
_, err = df.f.ReadAt(bytes, offset)
if err != nil {
if err == io.EOF {
df.fmutex.RUnlock()
continue
}
df.fmutex.RUnlock()
return
}
d = bytes
df.fmutex.RUnlock()
return
}
}
func (df *myDataFile) Write(d Data) (wsn int64, err error) {
// 读取并更新写偏移量。
var offset int64
df.wmutex.Lock()
offset = df.woffset
df.woffset += int64(df.dataLen)
df.wmutex.Unlock()
//写入一个数据块。
wsn = offset / int64(df.dataLen)
var bytes []byte
if len(d) > int(df.dataLen) {
bytes = d[0:df.dataLen]
} else {
bytes = d
}
df.fmutex.Lock()
defer df.fmutex.Unlock()
_, err = df.f.Write(bytes)
return
}
func (df *myDataFile) RSN() int64 {
df.rmutex.Lock()
defer df.rmutex.Unlock()
return df.roffset / int64(df.dataLen)
}
func (df *myDataFile) WSN() int64 {
df.wmutex.Lock()
defer df.wmutex.Unlock()
return df.woffset / int64(df.dataLen)
}
func (df *myDataFile) DataLen() uint32 {
return df.dataLen
}
func (df *myDataFile) Close() error {
if df.f == nil {
return nil
}
return df.f.Close()
}
datafile1_test.go：
package v1
import (
"math/rand"
"os"
"path/filepath"
"sync"
"testing"
)
func removeFile(path string) error {
file, err := os.Open(path)
if err != nil {
if !os.IsNotExist(err) {
return err
}
return nil
}
file.Close()
return os.Remove(path)
}
func TestIDataFile(t *testing.T) {
t.Run("v1/all", func(t *testing.T) {
dataLen := uint32(3)
path2 := filepath.Join(os.TempDir(), "data_file_test_new.txt")
defer func() {
if err := removeFile(path2); err != nil {
t.Errorf("Open file error: %s\n", err)
}
}()
t.Run("New", func(t *testing.T) {
testNew(path2, dataLen, t)
})
path3 := filepath.Join(os.TempDir(), "data_file_test.txt")
defer func() {
if err := removeFile(path3); err != nil {
t.Fatalf("Open file error: %s\n", err)
}
}()
max := 100000
t.Run("WriteAndRead", func(t *testing.T) {
testRW(path3, dataLen, max, t)
})
})
}
func testNew(path string, dataLen uint32, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
if dataFile == nil {
t.Log("Unnormal data file!")
t.FailNow()
}
defer dataFile.Close()
if dataFile.DataLen() != dataLen {
t.Fatalf("Incorrect data length!")
}
}
func testRW(path string, dataLen uint32, max int, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
defer dataFile.Close()
var wg sync.WaitGroup
wg.Add(5)
// 写入。
for i := 0; i < 3; i++ {
go func() {
defer wg.Done()
var prevWSN int64 = -1
for j := 0; j < max; j++ {
data := Data{
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
}
wsn, err := dataFile.Write(data)
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if prevWSN >= 0 && wsn <= prevWSN {
t.Fatalf("Incorect WSN %d! (lt %d)\n", wsn, prevWSN)
}
prevWSN = wsn
}
}()
}
// 读取。
for i := 0; i < 2; i++ {
go func() {
defer wg.Done()
var prevRSN int64 = -1
for i := 0; i < max; i++ {
rsn, date, err := dataFile.Read()
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if date == nil {
t.Fatalf("Unnormal data!")
}
if prevRSN >= 0 && rsn <= prevRSN {
t.Fatalf("Incorect RSN %d! (lt %d)\n", rsn, prevRSN)
}
prevRSN = rsn
}
}()
}
wg.Wait()
}
版本二：
datafile2.go：
package v2
import (
"errors"
"io"
"os"
"sync"
)
// Data 代表数据的类型。
type Data []byte
// DataFile 代表数据文件的接口类型。
type DataFile interface {
// Read 会读取一个数据块。
Read() (rsn int64, d Data, err error)
// Write 会写入一个数据块。
Write(d Data) (wsn int64, err error)
// RSN 会获取最后读取的数据块的序列号。
RSN() int64
// WSN 会获取最后写入的数据块的序列号。
WSN() int64
// DataLen 会获取数据块的长度。
DataLen() uint32
// Close 会关闭数据文件。
Close() error
}
// myDataFile 代表数据文件的实现类型。
type myDataFile struct {
f *os.File // 文件。
fmutex sync.RWMutex // 被用于文件的读写锁。
rcond *sync.Cond //读操作需要用到的条件变量
woffset int64 // 写操作需要用到的偏移量。
roffset int64 // 读操作需要用到的偏移量。
wmutex sync.Mutex // 写操作需要用到的互斥锁。
rmutex sync.Mutex // 读操作需要用到的互斥锁。
dataLen uint32 // 数据块长度。
}
// NewDataFile 会新建一个数据文件的实例。
func NewDataFile(path string, dataLen uint32) (DataFile, error) {
f, err := os.Create(path)
if err != nil {
return nil, err
}
if dataLen == 0 {
return nil, errors.New("Invalid data length!")
}
df := &myDataFile{f: f, dataLen: dataLen}
df.rcond = sync.NewCond(df.fmutex.RLocker())
return df, nil
}
func (df *myDataFile) Read() (rsn int64, d Data, err error) {
// 读取并更新读偏移量。
var offset int64
df.rmutex.Lock()
offset = df.roffset
df.roffset += int64(df.dataLen)
df.rmutex.Unlock()
//读取一个数据块。
rsn = offset / int64(df.dataLen)
bytes := make([]byte, df.dataLen)
df.fmutex.RLock()
defer df.fmutex.RUnlock()
for {
_, err = df.f.ReadAt(bytes, offset)
if err != nil {
if err == io.EOF {
df.rcond.Wait()
continue
}
return
}
d = bytes
return
}
}
func (df *myDataFile) Write(d Data) (wsn int64, err error) {
// 读取并更新写偏移量。
var offset int64
df.wmutex.Lock()
offset = df.woffset
df.woffset += int64(df.dataLen)
df.wmutex.Unlock()
//写入一个数据块。
wsn = offset / int64(df.dataLen)
var bytes []byte
if len(d) > int(df.dataLen) {
bytes = d[0:df.dataLen]
} else {
bytes = d
}
df.fmutex.Lock()
defer df.fmutex.Unlock()
_, err = df.f.Write(bytes)
df.rcond.Signal()
return
}
func (df *myDataFile) RSN() int64 {
df.rmutex.Lock()
defer df.rmutex.Unlock()
return df.roffset / int64(df.dataLen)
}
func (df *myDataFile) WSN() int64 {
df.wmutex.Lock()
defer df.wmutex.Unlock()
return df.woffset / int64(df.dataLen)
}
func (df *myDataFile) DataLen() uint32 {
return df.dataLen
}
func (df *myDataFile) Close() error {
if df.f == nil {
return nil
}
return df.f.Close()
}
datafile2_test.go：
package v2
import (
"math/rand"
"os"
"path/filepath"
"sync"
"testing"
)
func removeFile(path string) error {
file, err := os.Open(path)
if err != nil {
if !os.IsNotExist(err) {
return err
}
return nil
}
file.Close()
return os.Remove(path)
}
func TestIDataFile(t *testing.T) {
t.Run("v2/all", func(t *testing.T) {
dataLen := uint32(3)
path2 := filepath.Join(os.TempDir(), "data_file_test_new.txt")
defer func() {
if err := removeFile(path2); err != nil {
t.Errorf("Open file error: %s\n", err)
}
}()
t.Run("New", func(t *testing.T) {
testNew(path2, dataLen, t)
})
path3 := filepath.Join(os.TempDir(), "data_file_test.txt")
defer func() {
if err := removeFile(path3); err != nil {
t.Fatalf("Open file error: %s\n", err)
}
}()
max := 100000
t.Run("WriteAndRead", func(t *testing.T) {
testRW(path3, dataLen, max, t)
})
})
}
func testNew(path string, dataLen uint32, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
if dataFile == nil {
t.Log("Unnormal data file!")
t.FailNow()
}
defer dataFile.Close()
if dataFile.DataLen() != dataLen {
t.Fatalf("Incorrect data length!")
}
}
func testRW(path string, dataLen uint32, max int, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
defer dataFile.Close()
var wg sync.WaitGroup
wg.Add(5)
// 写入。
for i := 0; i < 3; i++ {
go func() {
defer wg.Done()
var prevWSN int64 = -1
for j := 0; j < max; j++ {
data := Data{
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
}
wsn, err := dataFile.Write(data)
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if prevWSN >= 0 && wsn <= prevWSN {
t.Fatalf("Incorect WSN %d! (lt %d)\n", wsn, prevWSN)
}
prevWSN = wsn
}
}()
}
// 读取。
for i := 0; i < 2; i++ {
go func() {
defer wg.Done()
var prevRSN int64 = -1
for i := 0; i < max; i++ {
rsn, date, err := dataFile.Read()
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if date == nil {
t.Fatalf("Unnormal data!")
}
if prevRSN >= 0 && rsn <= prevRSN {
t.Fatalf("Incorect RSN %d! (lt %d)\n", rsn, prevRSN)
}
prevRSN = rsn
}
}()
}
wg.Wait()
}版本三
datafile3.go：
package v3
import (
"errors"
"io"
"os"
"sync"
"sync/atomic"
)
// Data 代表数据的类型。
type Data []byte
// DataFile 代表数据文件的接口类型。
type DataFile interface {
// Read 会读取一个数据块。
Read() (rsn int64, d Data, err error)
// Write 会写入一个数据块。
Write(d Data) (wsn int64, err error)
// RSN 会获取最后读取的数据块的序列号。
RSN() int64
// WSN 会获取最后写入的数据块的序列号。
WSN() int64
// DataLen 会获取数据块的长度。
DataLen() uint32
// Close 会关闭数据文件。
Close() error
}
// myDataFile 代表数据文件的实现类型。
type myDataFile struct {
f *os.File // 文件。
fmutex sync.RWMutex // 被用于文件的读写锁。
rcond *sync.Cond // 读操作需要用到的条件变量
woffset int64 // 写操作需要用到的偏移量。
roffset int64 // 读操作需要用到的偏移量。
dataLen uint32 // 数据块长度。
}
// NewDataFile 会新建一个数据文件的实例。
func NewDataFile(path string, dataLen uint32) (DataFile, error) {
f, err := os.Create(path)
if err != nil {
return nil, err
}
if dataLen == 0 {
return nil, errors.New("Invalid data length!")
}
df := &myDataFile{f: f, dataLen: dataLen}
df.rcond = sync.NewCond(df.fmutex.RLocker())
return df, nil
}
func (df *myDataFile) Read() (rsn int64, d Data, err error) {
// 读取并更新读偏移量
var offset int64
for {
offset = atomic.LoadInt64(&df.roffset)
if atomic.CompareAndSwapInt64(&df.roffset, offset, (offset + int64(df.dataLen))) {
break
}
}
//读取一个数据块
rsn = offset / int64(df.dataLen)
bytes := make([]byte, df.dataLen)
df.fmutex.RLock()
defer df.fmutex.RUnlock()
for {
_, err = df.f.ReadAt(bytes, offset)
if err != nil {
if err == io.EOF {
df.rcond.Wait()
continue
}
return
}
d = bytes
return
}
}
func (df *myDataFile) Write(d Data) (wsn int64, err error) {
// 读取并更新写偏移量
var offset int64
for {
offset = atomic.LoadInt64(&df.woffset)
if atomic.CompareAndSwapInt64(&df.woffset, offset, (offset + int64(df.dataLen))) {
break
}
}
//写入一个数据块
wsn = offset / int64(df.dataLen)
var bytes []byte
if len(d) > int(df.dataLen) {
bytes = d[0:df.dataLen]
} else {
bytes = d
}
df.fmutex.Lock()
defer df.fmutex.Unlock()
_, err = df.f.Write(bytes)
df.rcond.Signal()
return
}
func (df *myDataFile) RSN() int64 {
offset := atomic.LoadInt64(&df.roffset)
return offset / int64(df.dataLen)
}
func (df *myDataFile) WSN() int64 {
offset := atomic.LoadInt64(&df.woffset)
return offset / int64(df.dataLen)
}
func (df *myDataFile) DataLen() uint32 {
return df.dataLen
}
func (df *myDataFile) Close() error {
if df.f == nil {
return nil
}
return df.f.Close()
}
datafile3_test.go：
package v3
import (
"math/rand"
"os"
"path/filepath"
"sync"
"testing"
)
func removeFile(path string) error {
file, err := os.Open(path)
if err != nil {
if !os.IsNotExist(err) {
return err
}
return nil
}
file.Close()
return os.Remove(path)
}
func TestIDataFile(t *testing.T) {
t.Run("v3/all", func(t *testing.T) {
dataLen := uint32(3)
path2 := filepath.Join(os.TempDir(), "data_file_test_new.txt")
defer func() {
if err := removeFile(path2); err != nil {
t.Errorf("Open file error: %s\n", err)
}
}()
t.Run("New", func(t *testing.T) {
testNew(path2, dataLen, t)
})
path3 := filepath.Join(os.TempDir(), "data_file_test.txt")
defer func() {
if err := removeFile(path3); err != nil {
t.Fatalf("Open file error: %s\n", err)
}
}()
max := 100000
t.Run("WriteAndRead", func(t *testing.T) {
testRW(path3, dataLen, max, t)
})
})
}
func testNew(path string, dataLen uint32, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
if dataFile == nil {
t.Log("Unnormal data file!")
t.FailNow()
}
defer dataFile.Close()
if dataFile.DataLen() != dataLen {
t.Fatalf("Incorrect data length!")
}
}
func testRW(path string, dataLen uint32, max int, t *testing.T) {
t.Logf("New a data file (path: %s, dataLen: %d)...\n",
path, dataLen)
dataFile, err := NewDataFile(path, dataLen)
if err != nil {
t.Logf("Couldn't new a data file: %s", err)
t.FailNow()
}
defer dataFile.Close()
var wg sync.WaitGroup
wg.Add(5)
// 写入。
for i := 0; i < 3; i++ {
go func() {
defer wg.Done()
var prevWSN int64 = -1
for j := 0; j < max; j++ {
data := Data{
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
byte(rand.Int31n(256)),
}
wsn, err := dataFile.Write(data)
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if prevWSN >= 0 && wsn <= prevWSN {
t.Fatalf("Incorect WSN %d! (lt %d)\n", wsn, prevWSN)
}
prevWSN = wsn
}
}()
}
// 读取。
for i := 0; i < 2; i++ {
go func() {
defer wg.Done()
var prevRSN int64 = -1
for i := 0; i < max; i++ {
rsn, date, err := dataFile.Read()
if err != nil {
t.Fatalf("Unexpected writing error: %s\n", err)
}
if date == nil {
t.Fatalf("Unnormal data!")
}
if prevRSN >= 0 && rsn <= prevRSN {
t.Fatalf("Incorect RSN %d! (lt %d)\n", rsn, prevRSN)
}
prevRSN = rsn
}
}()
}
wg.Wait()
}
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
