golib

README

golib

介绍

{以下是码云平台说明，您可以替换此简介 码云是 OSCHINA 推出的基于 Git 的代码托管平台（同时支持 SVN）。专为开发者提供稳定、高效、安全的云端软件开发协作平台 无论是个人、团队、或是企业，都能够用码云实现代码托管、项目管理、协作开发。企业项目请看 https://gitee.com/enterprises}

软件架构

软件架构说明

安装教程

1. xxxx
2. xxxx
3. xxxx

使用说明

1. xxxx
2. xxxx
3. xxxx

参与贡献

1. Fork 本仓库
2. 新建 Feat_xxx 分支
3. 提交代码
4. 新建 Pull Request

码云特技

1. 使用 Readme_XXX.md 来支持不同的语言，例如 Readme_en.md, Readme_zh.md
2. 码云官方博客 blog.gitee.com
3. 你可以 https://gitee.com/explore 这个地址来了解码云上的优秀开源项目
4. GVP 全称是码云最有价值开源项目，是码云综合评定出的优秀开源项目
5. 码云官方提供的使用手册 https://gitee.com/help
6. 码云封面人物是一档用来展示码云会员风采的栏目 https://gitee.com/gitee-stars/

Documentation

Overview

* Kafka操作封装 * @author yorkershi * @created 2020-2-28

Index

• type Cache
  • func NewCache() *Cache
  • func (this *Cache) Clean()
  • func (this *Cache) Get(key string) interface{}
  • func (this *Cache) GetInt(key string) int64
  • func (this *Cache) GetMap(key string) map[string]string
  • func (this *Cache) GetMapInterface(key string) map[string]interface{}
  • func (this *Cache) GetMapInterfaceList(key string) []map[string]interface{}
  • func (this *Cache) GetString(key string) string
  • func (this *Cache) GetStringList(key string) []string
  • func (this *Cache) Set(key string, value interface{}, expiration time.Duration)
  • func (this *Cache) SetInt(key string, value int64, expiration time.Duration)
  • func (this *Cache) SetMap(key string, value map[string]string, expiration time.Duration)
  • func (this *Cache) SetMapInterface(key string, value map[string]interface{}, expiration time.Duration)
  • func (this *Cache) SetMapInterfaceList(key string, value []map[string]interface{}, expiration time.Duration)
  • func (this *Cache) SetString(key, value string, expiration time.Duration)
  • func (this *Cache) SetStringList(key string, value []string, expiration time.Duration)
• type Kafka
  • func NewKafka(topic string, partition int) (*Kafka, error)
  • func (this *Kafka) Close()
  • func (this *Kafka) LookupPartitions() ([]kafka.Partition, error)
  • func (this *Kafka) ReadMessage() ([]byte, error)
  • func (this *Kafka) ReadMessages() ([][]byte, error)
  • func (this *Kafka) SetReadOffset(offset int64) error
  • func (this *Kafka) SetReadOffsetByTime(tm time.Time) error
  • func (this *Kafka) WriteMessage(msg []byte) (int, error)
  • func (this *Kafka) WriteMessages(msglist [][]byte) (int, error)

Types

type Cache

type Cache struct {
}

func NewCache

func NewCache() *Cache

func (*Cache) Clean

func (this *Cache) Clean()

定时清理过期的缓存数据

func (*Cache) Get

func (this *Cache) Get(key string) interface{}

从cache中获取一个值

func (*Cache) GetInt

func (this *Cache) GetInt(key string) int64

func (*Cache) GetMap

func (this *Cache) GetMap(key string) map[string]string

func (*Cache) GetMapInterface

func (this *Cache) GetMapInterface(key string) map[string]interface{}

func (*Cache) GetMapInterfaceList

func (this *Cache) GetMapInterfaceList(key string) []map[string]interface{}

func (*Cache) GetString

func (this *Cache) GetString(key string) string

func (*Cache) GetStringList

func (this *Cache) GetStringList(key string) []string

func (*Cache) Set

func (this *Cache) Set(key string, value interface{}, expiration time.Duration)

设置一个值到cache中

func (*Cache) SetInt

func (this *Cache) SetInt(key string, value int64, expiration time.Duration)

设置一个int64类型到cache中

func (*Cache) SetMap

func (this *Cache) SetMap(key string, value map[string]string, expiration time.Duration)

设置一个map[string]string内型到cache中

func (*Cache) SetMapInterface

func (this *Cache) SetMapInterface(key string, value map[string]interface{}, expiration time.Duration)

设置一个map[string]interface{}内型到cache中

func (*Cache) SetMapInterfaceList

func (this *Cache) SetMapInterfaceList(key string, value []map[string]interface{}, expiration time.Duration)

设置一个[]map[string]interface{}内型到cache中

func (*Cache) SetString

func (this *Cache) SetString(key, value string, expiration time.Duration)

设置一个字符串到cache中

func (*Cache) SetStringList

func (this *Cache) SetStringList(key string, value []string, expiration time.Duration)

设置一个字符串列表到Cache中

type Kafka

type Kafka struct {
	Conn *kafka.Conn
	// contains filtered or unexported fields
}

func NewKafka

func NewKafka(topic string, partition int) (*Kafka, error)

实例化Kafka操作类

func (*Kafka) Close

func (this *Kafka) Close()

关闭kafka连接

func (*Kafka) LookupPartitions

func (this *Kafka) LookupPartitions() ([]kafka.Partition, error)

获取topic对象的所有分区信息, 拿到分区的总数量和下标位置后,可以再次实现化特别分区的Kafka对象进行操作

func (*Kafka) ReadMessage

func (this *Kafka) ReadMessage() ([]byte, error)

从当前位置开始读取一条记录

func (*Kafka) ReadMessages

func (this *Kafka) ReadMessages() ([][]byte, error)

从当前位置开始批量读取数据

func (*Kafka) SetReadOffset

func (this *Kafka) SetReadOffset(offset int64) error

设置从起点位置开始的偏移量

func (*Kafka) SetReadOffsetByTime

func (this *Kafka) SetReadOffsetByTime(tm time.Time) error

设置从某个时间开始的偏移量

func (*Kafka) WriteMessage

func (this *Kafka) WriteMessage(msg []byte) (int, error)

将一条消息写入到kafka中

func (*Kafka) WriteMessages

func (this *Kafka) WriteMessages(msglist [][]byte) (int, error)

将一组消息写入到kafka的某一个topic中的某一个partition中, 完成写入后连接关闭, 返回完成写入的字节数