撸一个预言机（Oracle）服务，真香！—中篇

六天
更新于 2020-06-18 14:50
阅读 5101

本文将通过上、中、下三篇文章带领大家一步步开发实现一个自己中心化的Oracle服务，并通过抽奖合约演示如何使用。

# 一、文章结构
本文将通过上、中、下三篇文章带领大家一步步开发实现一个中心化的Oracle服务，并通过一个抽奖合约演示如何使用我们的Oracle服务。文章内容安排如下：
- 上篇：[Oracle简介及合约实现](https://learnblockchain.cn/article/1150)
- 中篇：**使用go语言开发Oracle服务**
- 下篇：[抽奖合约调用Oracle服务示例](https://learnblockchain.cn/article/1162)

在上篇中，我们实现了一个通用的Oracle合约，其主要有一个接收用户请求的Query方法；回调用户合约的Response方法和一个供Oracle后端服务订阅的QueryInfo事件。

本篇是中篇，主要使用go语言开发实现Oracle的后端服务。

> 本文作者六天，文中的Oracle服务完整代码地址：[https://github.com/six-days/ethereum-oracle-service](https://github.com/six-days/ethereum-oracle-service)

# 二、服务架构

Oracle后端服务整体包含**事件订阅模块**、**查询模块**和**回调模块**，架构如下图所示。

![架构图](https://img.learnblockchain.cn/attachments/2020/06/mWmTZPHr5eeb0e10512c1.png)

服务开启后，首先会通过以太坊**ws协议的jsonrpc**，在区块链上注册事件订阅，订阅成功后开启一个for循环，接收并处理事件消息。

代码如下所示。

```
// start monitor oracle contract event
func (e *EventWatch) Start() {
	if err := e.subscribeEvent(); err != nil {
		return
	}
	e.dealEvent()
}

func (e *EventWatch) dealEvent() {
	for {
		select {
		case err := <-e.Subscription.Err():
			logs.Error("[dealEvent] Subscription err: ", err)
			e.subscribeEvent()
		case vLog := <-e.EventChan:
			// 处理查询请求并回调
			go e.dealQuery(vLog)
		}
	}
}
```

# 三、事件订阅
**事件订阅必须使用ws协议的jsonrpc，http协议的jsonprc无法订阅事件。**

事件订阅的核心是通过ethclient的`SubscribeFilterLogs`方法，其中`query`参数是订阅的过滤条件。其中
- `Addresses `是Oracle合约地址；
- `Topics `参数是过滤主题，是一个二维数组，这里我们的主题只指定了事件的名称。

代码如下所示。
```
func (e *EventWatch) subscribeEvent() error {
	query := ethereum.FilterQuery{
		Addresses: []common.Address{
			common.HexToAddress(e.Config.OracleContractAddress),
		},
		Topics: [][]common.Hash{
			{e.OracleABI.Events[OracelEventName].ID()},
		},
	}
	events := make(chan types.Log)
	sub, err := e.Client.SubscribeFilterLogs(context.Background(), query, events)
	if err != nil {
		logs.Error("[SubscribeEvent]fail to subscribe event:", err)
		return err
	}
	e.EventChan = events
	e.Subscription = sub
	return nil
}
```

# 四、查询模块
## 1、日志解析
事件日志解析我们用go-ethereum的abi模块的`Unpack`方法，将日志解析为我们定义好的结构体。

代码如下所示。
```
type OracleQueryInfo struct {
	QueryId      [32]byte
	Requester    common.Address
	Fee          *big.Int
	CallbackAddr common.Address
	CallbackFUN  string
	QueryData    []byte
	Raw          types.Log // Blockchain specific contextual infos
}

type QueryRequest struct {
	URL            string   `json:"url,omitempty"`
	ResponseParams []string `json:"responseParams,omitempty"`
}

func (e *EventWatch) dealQuery(vLog types.Log) error {
        queryInfo := &OracleQueryInfo{}
	err := e.OracleABI.Unpack(queryInfo, OracelEventName, vLog.Data)
	if err != nil {
		return fmt.Errorf("[dealQuery] unpack event log failed:%v", err)
	}
	reqData := &QueryRequest{}
	if err = json.Unmarshal(queryInfo.QueryData, reqData); err != nil {
		return fmt.Errorf("[dealQuery] unmarshal query data failed:%v", err)
	}
}
```
## 2、查询请求
查询请求比较简单，就是根据用户提供的url发送请求。代码如下所示。

```
// sendQueryRequest 根据客户端指定的查询地址发送请求
func (e *EventWatch) sendQueryRequest(reqData *QueryRequest, resParamType string) (interface{}, error) {
	req, err := http.NewRequest("GET", reqData.URL, nil)
	if err != nil {
		return nil, fmt.Errorf("[sendQueryRequest] NewRequest failed: %v", err)
	}
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("[sendQueryRequest] http get request failed: %v", err)
	}
	defer res.Body.Close()
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return nil, fmt.Errorf("[sendQueryRequest] read response data failed: %v", err)
	}
	logs.Trace("[sendQueryRequest] get ", reqData.URL, " response is: ", string(body))
	queryRes, err := ParseResponeData(body, reqData.ResponseParams, resParamType)
	if err != nil {
		return nil, err
	}
	return queryRes, nil
}
```
查询可能失败，这里需要增加失败重试机制，代码比较简单，就不写出来了。

## 3、结果解析

这里使用`go-simplejson`库将查询结果进行json解析，并且提取用户指定所需要的字段，将字段转换为用户合约中回调方法接收的数据类型。

```
// ParseResponeData 解析链下获取到的数据，提取用户所需要的字段，并转换为对应的数据类型
func ParseResponeData(repData []byte, keys []string, resParamType string) (interface{}, error) {
	resData, err := simplejson.NewJson(repData)
	if err != nil {
		return nil, fmt.Errorf("[ParseResponeData] unmarshal response data failed:%v", err)
	}

for _, paramName := range keys {
		resData = resData.Get(paramName)
	}
	if resData == nil {
		return nil, fmt.Errorf("[ParseResponeData] response data not exist request key:%v", keys)
	}
	var resValue interface{}
	var coverErr error
	switch resParamType {
	case "uint256":
		resUint64Value, coverErr := resData.Uint64()
		if coverErr == nil {
			resValue = big.NewInt(int64(resUint64Value))
		}
	
	case "bytes":
		resValue, coverErr = resData.Bytes()
	default:
		return nil, fmt.Errorf("[ParseResponeData] unsupport response data type %s", resParamType)
	}
	if coverErr != nil {
		return nil, fmt.Errorf("[ParseResponeData] response data type %s error:%v", resParamType, err)
	}

return resValue, nil
}
```

# 五、回调模块
回调模块相对比较简单，首先将Oracle合约实例化了一个`BoundContract`对象，然后调用`Transact`方法发送交易。其中第一个参数是使用私钥实例化的一个`TransactOpts`对象。

在`TransactOpts`对象中可以配置nonce、gasLimit、gasPrice等值，如果不指定，`Transact`方法会自己补充上。除此之外，`Transact`方法也会调用`TransactOpts`对象的`Signer`方法对消息进行签名。

`Transact`方法源码详见：[https://github.com/six-days/go-ethereum/blob/master/accounts/abi/bind/base.go](https://github.com/six-days/go-ethereum/blob/master/accounts/abi/bind/base.go)

回调模块代码如下所示。
```
// sendQueryResponse 将查询到的结果发送给客户端合约指定方法
func (e *EventWatch) sendQueryResponse(res interface{}, stateCode uint64, queryInfo *OracleQueryInfo, resParamType string) error {
	in := []interface{}{
		queryInfo.QueryId,
		queryInfo.CallbackAddr,
		queryInfo.CallbackFUN,
		stateCode,
		res,
	}
	var responseName string
	switch resParamType {
	case "bytes":
		responseName = OracelResponseBytesName
	case "uint256":
		responseName = OracelResponseUint256Name
	default:
		return fmt.Errorf("[SendQueryResponse] unsupport response data type")
	}
	transaction, err := e.BoundContract.Transact(e.TransactOpts, responseName, in...)
	if err != nil {
		return fmt.Errorf("[SendQueryResponse] Transact failed: %v", err)
	}
	logs.Trace("[SendQueryResponse] call back tx:", transaction.Hash().Hex())
	return nil
}
```
回调也可能失败，服务对`sendQueryResponse `方法的调用也增加了失败重试机制。

# 六、编译与运行

## 1、编译
```
go build
```
编译完成后查看帮助信息

```
./oracle-service -h
oracle_service version: 1.0.0
Usage: oracle_service [-h help] [-v version] [-c config path] [-l log path]
```
## 2、配置

配置信息如下：

```
# 合约地址
OracleContractAddress = ""
# 网络ws地址
NetworkWS = "ws://"
# 调用合约的私钥
PrivateKey = ""
```

## 3、运行
```
./oracle-service -c ./conf/app.conf -l logs/
```

# 七、可以优化的地方
至此，我们的V1版的Oracle服务已开发完成，服务已能满足基本需求，但还有一些方面需要进一步优化，我这里列出了三点。

## 1、Nonce托管
在回调模块中，调用合约时，我们并没有指定发起交易账号的Nonce值，而是由`Transact`方法在每次发起交易时，动态计算。这就会限制我们交易的并发。

在高并发的情况下，肯定会出现多笔交易Nonce值相同的情况，后发起交易覆盖前交易，造成前交易失败。

针对这种情况，我的思路是对Nonce进行托管：
- 在缓存（内存或redis等）中维护账号对应的Nonce
- 每次发起交易时，从缓存中获取，每获取一次，缓存中的Nonce累加1
- 缓存中的Nonce定期和链上进行校对和同步
- 对于可能出现的空洞情况，使用空交易填补

## 2、Gas优化
这段时间以太坊网络比较拥堵，导致手续费居高不下。对于我们Oracle服务来说，节省Gas是很重要的一个优化方向。

这里我的思路是可以从以下几个方面优化：
- 引入动态GasPrice，可以从https://ethgasstation.info网站中获取实时的GasPrice
- 指定GasLimit，防止由于合约问题消耗过多Gas
- 余额检查，防止由于余额不足造成交易失败，浪费了手续费
- 接收回调数据的用户合约方法尽量简单，分离业务逻辑

## 3、支持http协议jsonrpc
有的网络节点没有开启ws服务，而使用http协议的网络jsonrpc又无法直接订阅事件。这时可以采取迂回策略，模拟事件订阅，具体思路如下：
- 开启网络区块监控
- 监控到有新区块产生，查询区块中的日志
- 如果有我们Oracle合约产生的查询日志，则进入后续的查询和回调流程

大家对于优化有其他思路或疑问，欢迎留言探讨。

**下篇中，我将以一个抽奖合约为示例，介绍如何使用我们开发的Oracle服务来对抽奖合约提供一个随机数。**

一、文章结构

本文将通过上、中、下三篇文章带领大家一步步开发实现一个中心化的Oracle服务，并通过一个抽奖合约演示如何使用我们的Oracle服务。文章内容安排如下：

上篇：Oracle简介及合约实现
中篇：使用go语言开发Oracle服务
下篇：抽奖合约调用Oracle服务示例

本篇是中篇，主要使用go语言开发实现Oracle的后端服务。

本文作者六天，文中的Oracle服务完整代码地址：https://github.com/six-days/ethereum-oracle-service

二、服务架构

Oracle后端服务整体包含事件订阅模块、查询模块和回调模块，架构如下图所示。

服务开启后，首先会通过以太坊ws协议的jsonrpc，在区块链上注册事件订阅，订阅成功后开启一个for循环，接收并处理事件消息。

代码如下所示。

// start monitor oracle contract event
func (e *EventWatch) Start() {
    if err := e.subscribeEvent(); err != nil {
        return
    }
    e.dealEvent()
}

func (e *EventWatch) dealEvent() {
    for {
        select {
        case err := &lt;-e.Subscription.Err():
            logs.Error("[dealEvent] Subscription err: ", err)
            e.subscribeEvent()
        case vLog := &lt;-e.EventChan:
            // 处理查询请求并回调
            go e.dealQuery(vLog)
        }
    }
}

三、事件订阅

事件订阅必须使用ws协议的jsonrpc，http协议的jsonprc无法订阅事件。

事件订阅的核心是通过ethclient的SubscribeFilterLogs方法，其中query参数是订阅的过滤条件。其中

Addresses是Oracle合约地址；
Topics参数是过滤主题，是一个二维数组，这里我们的主题只指定了事件的名称。

代码如下所示。

func (e *EventWatch) subscribeEvent() error {
    query := ethereum.FilterQuery{
        Addresses: []common.Address{
            common.HexToAddress(e.Config.OracleContractAddress),
        },
        Topics: [][]common.Hash{
            {e.OracleABI.Events[OracelEventName].ID()},
        },
    }
    events := make(chan types.Log)
    sub, err := e.Client.SubscribeFilterLogs(context.Background(), query, events)
    if err != nil {
        logs.Error("[SubscribeEvent]fail to subscribe event:", err)
        return err
    }
    e.EventChan = events
    e.Subscription = sub
    return nil
}

四、查询模块

1、日志解析

事件日志解析我们用go-ethereum的abi模块的Unpack方法，将日志解析为我们定义好的结构体。

代码如下所示。

type OracleQueryInfo struct {
    QueryId      [32]byte
    Requester    common.Address
    Fee          *big.Int
    CallbackAddr common.Address
    CallbackFUN  string
    QueryData    []byte
    Raw          types.Log // Blockchain specific contextual infos
}

type QueryRequest struct {
    URL            string   `json:"url,omitempty"`
    ResponseParams []string `json:"responseParams,omitempty"`
}

func (e *EventWatch) dealQuery(vLog types.Log) error {
        queryInfo := &OracleQueryInfo{}
    err := e.OracleABI.Unpack(queryInfo, OracelEventName, vLog.Data)
    if err != nil {
        return fmt.Errorf("[dealQuery] unpack event log failed:%v", err)
    }
    reqData := &QueryRequest{}
    if err = json.Unmarshal(queryInfo.QueryData, reqData); err != nil {
        return fmt.Errorf("[dealQuery] unmarshal query data failed:%v", err)
    }
}

2、查询请求

查询请求比较简单，就是根据用户提供的url发送请求。代码如下所示。

// sendQueryRequest 根据客户端指定的查询地址发送请求
func (e *EventWatch) sendQueryRequest(reqData *QueryRequest, resParamType string) (interface{}, error) {
    req, err := http.NewRequest("GET", reqData.URL, nil)
    if err != nil {
        return nil, fmt.Errorf("[sendQueryRequest] NewRequest failed: %v", err)
    }
    res, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("[sendQueryRequest] http get request failed: %v", err)
    }
    defer res.Body.Close()
    body, err := ioutil.ReadAll(res.Body)
    if err != nil {
        return nil, fmt.Errorf("[sendQueryRequest] read response data failed: %v", err)
    }
    logs.Trace("[sendQueryRequest] get ", reqData.URL, " response is: ", string(body))
    queryRes, err := ParseResponeData(body, reqData.ResponseParams, resParamType)
    if err != nil {
        return nil, err
    }
    return queryRes, nil
}

查询可能失败，这里需要增加失败重试机制，代码比较简单，就不写出来了。

3、结果解析

这里使用go-simplejson库将查询结果进行json解析，并且提取用户指定所需要的字段，将字段转换为用户合约中回调方法接收的数据类型。

// ParseResponeData 解析链下获取到的数据，提取用户所需要的字段，并转换为对应的数据类型
func ParseResponeData(repData []byte, keys []string, resParamType string) (interface{}, error) {
    resData, err := simplejson.NewJson(repData)
    if err != nil {
        return nil, fmt.Errorf("[ParseResponeData] unmarshal response data failed:%v", err)
    }

    for _, paramName := range keys {
        resData = resData.Get(paramName)
    }
    if resData == nil {
        return nil, fmt.Errorf("[ParseResponeData] response data not exist request key:%v", keys)
    }
    var resValue interface{}
    var coverErr error
    switch resParamType {
    case "uint256":
        resUint64Value, coverErr := resData.Uint64()
        if coverErr == nil {
            resValue = big.NewInt(int64(resUint64Value))
        }

    case "bytes":
        resValue, coverErr = resData.Bytes()
    default:
        return nil, fmt.Errorf("[ParseResponeData] unsupport response data type %s", resParamType)
    }
    if coverErr != nil {
        return nil, fmt.Errorf("[ParseResponeData] response data type %s error:%v", resParamType, err)
    }

    return resValue, nil
}

五、回调模块

回调模块相对比较简单，首先将Oracle合约实例化了一个BoundContract对象，然后调用Transact方法发送交易。其中第一个参数是使用私钥实例化的一个TransactOpts对象。

在TransactOpts对象中可以配置nonce、gasLimit、gasPrice等值，如果不指定，Transact方法会自己补充上。除此之外，Transact方法也会调用TransactOpts对象的Signer方法对消息进行签名。

Transact方法源码详见：https://github.com/six-days/go-ethereum/blob/master/accounts/abi/bind/base.go

回调模块代码如下所示。

// sendQueryResponse 将查询到的结果发送给客户端合约指定方法
func (e *EventWatch) sendQueryResponse(res interface{}, stateCode uint64, queryInfo *OracleQueryInfo, resParamType string) error {
    in := []interface{}{
        queryInfo.QueryId,
        queryInfo.CallbackAddr,
        queryInfo.CallbackFUN,
        stateCode,
        res,
    }
    var responseName string
    switch resParamType {
    case "bytes":
        responseName = OracelResponseBytesName
    case "uint256":
        responseName = OracelResponseUint256Name
    default:
        return fmt.Errorf("[SendQueryResponse] unsupport response data type")
    }
    transaction, err := e.BoundContract.Transact(e.TransactOpts, responseName, in...)
    if err != nil {
        return fmt.Errorf("[SendQueryResponse] Transact failed: %v", err)
    }
    logs.Trace("[SendQueryResponse] call back tx:", transaction.Hash().Hex())
    return nil
}

回调也可能失败，服务对sendQueryResponse方法的调用也增加了失败重试机制。

六、编译与运行

1、编译

go build

编译完成后查看帮助信息

./oracle-service -h
oracle_service version: 1.0.0
Usage: oracle_service [-h help] [-v version] [-c config path] [-l log path]

2、配置

配置信息如下：

# 合约地址
OracleContractAddress = ""
# 网络ws地址
NetworkWS = "ws://"
# 调用合约的私钥
PrivateKey = ""

3、运行

./oracle-service -c ./conf/app.conf -l logs/

七、可以优化的地方

至此，我们的V1版的Oracle服务已开发完成，服务已能满足基本需求，但还有一些方面需要进一步优化，我这里列出了三点。

1、Nonce托管

在回调模块中，调用合约时，我们并没有指定发起交易账号的Nonce值，而是由Transact方法在每次发起交易时，动态计算。这就会限制我们交易的并发。

在高并发的情况下，肯定会出现多笔交易Nonce值相同的情况，后发起交易覆盖前交易，造成前交易失败。

针对这种情况，我的思路是对Nonce进行托管：

在缓存（内存或redis等）中维护账号对应的Nonce
每次发起交易时，从缓存中获取，每获取一次，缓存中的Nonce累加1
缓存中的Nonce定期和链上进行校对和同步
对于可能出现的空洞情况，使用空交易填补

2、Gas优化

这段时间以太坊网络比较拥堵，导致手续费居高不下。对于我们Oracle服务来说，节省Gas是很重要的一个优化方向。

这里我的思路是可以从以下几个方面优化：

引入动态GasPrice，可以从https://ethgasstation.info网站中获取实时的GasPrice
指定GasLimit，防止由于合约问题消耗过多Gas
余额检查，防止由于余额不足造成交易失败，浪费了手续费
接收回调数据的用户合约方法尽量简单，分离业务逻辑

3、支持http协议jsonrpc

有的网络节点没有开启ws服务，而使用http协议的网络jsonrpc又无法直接订阅事件。这时可以采取迂回策略，模拟事件订阅，具体思路如下：

开启网络区块监控
监控到有新区块产生，查询区块中的日志
如果有我们Oracle合约产生的查询日志，则进入后续的查询和回调流程

大家对于优化有其他思路或疑问，欢迎留言探讨。

下篇中，我将以一个抽奖合约为示例，介绍如何使用我们开发的Oracle服务来对抽奖合约提供一个随机数。

学分: 198
分类: 以太坊
标签: 预言机事件订阅以太坊日志

本文参与登链社区写作激励计划，好文好收益，欢迎正在阅读的你也加入。

撸一个预言机（Oracle）服务，真香！—中篇

一、文章结构

二、服务架构

三、事件订阅

四、查询模块

1、日志解析

2、查询请求

3、结果解析

五、回调模块

六、编译与运行

1、编译

2、配置

3、运行

七、可以优化的地方

1、Nonce托管

2、Gas优化

3、支持http协议jsonrpc

1 条评论

文章目录