Arbitrum Nitro交易速度压力测试实战：TPS性能评估全解析

Arbitrum Nitro 是一种基于以太坊的 Layer 2 扩展解决方案，旨在提高交易吞吐量并降低交易费用。为了全面评估其性能，我们需要进行了详细的压力测试。本文的目的是回顾一下我在实际测试过程中采用的方法，还有测试的思路。

我们的压力测试主要目标是评估 Arbitrum Nitro 在高负载下的性能，包括每秒交易数（TPS）、交易确认时间等关键指标。

测试原理和机制

测试的原理是通过模拟大量交易请求来测试系统的处理能力。测试过程包括以下几个关键步骤：

1. 生成测试账户：创建多个钱包地址用于交易测试。
2. 转账测试：频繁进行转账操作，模拟真实交易场景。
3. 数据记录与分析：记录每笔交易的开始时间、确认时间和结束时间，并对数据进行分析。

在测试过程中，会记录请求信息，然后下载生成的区块，然后分析交易生成的数据，评估交易的平均处理时间、每秒钟平均处理交易笔数等指标。

代码和执行过程

文中代码只是为了说明测试思路，类似伪代码，实际执行的代码可以自己根据思路实现，也可以联系作者。

1. 生成测试账户和发起交易请求

使用 web3.js 库生成测试账户，并将它们保存到本地 JSON 文件中，同时记录交易数据到 JSONL 文件：

const { Web3 } = require('web3');
const fs = require('fs');
const web3 = new Web3('http://localhost:8545');

// 生成测试账户
const generateAccounts = async (numAccounts) => {
    const accounts = {};
    for (let i = 0; i < numAccounts; i++) {
        const account = web3.eth.accounts.create();
        accounts[account.address] = account.privateKey;
    }
    fs.writeFileSync('wallets.json', JSON.stringify(accounts, null, 2));
};

// 发送交易并记录数据
const sendTransaction = async (from, to, value, privateKey) => {
    const tx = {
        to,
        value,
        gas: 21000,
        nonce: await web3.eth.getTransactionCount(from),
    };
    const signedTx = await web3.eth.accounts.signTransaction(tx, privateKey);
    const startTime = Date.now();
    web3.eth.sendSignedTransaction(signedTx.rawTransaction)
        .on('receipt', (receipt) => {
            const responseTime = Date.now();
            const logEntry = {
                start_time: startTime,
                response_time: responseTime,
                tx_hash: receipt.transactionHash,
            };
            fs.appendFileSync('transactions.jsonl', JSON.stringify(logEntry) + '\n');
        });
};

// 主函数
const main = async () => {
    const accounts = JSON.parse(fs.readFileSync('accounts.json'));
    const mainAccount = '0xYourMainAccount';
    const mainPrivateKey = '0xYourMainPrivateKey';

    for (const [address, privateKey] of Object.entries(accounts)) {
        await sendTransaction(mainAccount, address, web3.utils.toWei('1', 'ether'), mainPrivateKey);
        await sendTransaction(address, mainAccount, web3.utils.toWei('0.001', 'ether'), privateKey);
    }
};

// 生成测试账户并执行主函数
generateAccounts(100).then(() => main());

通过设置要生成的测试账户数量，就可以简单的控制同时发起请求的数量。在发起交易的时候，我们可能会希望能够更精确控制每秒钟发起交易的数量，还有测试的持续时间，可以继续改造，增加并发控制。

另外，为了更好地控制测试速度，我们将转账分成两步，第一步先批量向生成的钱包中转入1-2个ETH，然后压力测试过程中，从生成的钱包地址，用小金额将ETH转回主账号。

改造之后代码如下：

const options = {  duration: 30, // 单位秒  number: 100,  tps: 200,  wait: true,}async function runStrssTest(accounts, options) {  const startAt = Date.now()  const ctrlDict = {}  while(true) {    if (Date.now() - startAt > options.duration * 1000) {      break    }    
    const sec = Math.floor(Date.now() / 1000)    if (!ctrlDict[sec]) {        ctrlDict[sec] = 0    }    for (const [address, privateKey] of Object.entries(accounts)) {        if (ctrlDict[sec] >= options.tps) {            continue        }        ctrlDict[sec]++        sendTransaction(address, mainAccount, web3.utils.toWei('0.001', 'ether'), privateKey);    }   }}async function main() {    ...    batchSendBalanceFromMain(mainPrivateKey, accounts)    ...    await runStrssTest(mainAccount, accounts, options)    ...}

并发批量下载区块数据并写入 JSONL 文件

使用 Python 并发下载区块数据，并将其保存到 JSONL 文件中：

import json
import time
import threading
from web3 import Web3
from concurrent.futures import ThreadPoolExecutor
web3 = Web3(Web3.HTTPProvider('http://localhost:8545'))
lock = threading.Lock()

def get_block(block_number):
    try:
        print('get_block', block_number)
        block = web3.eth.get_block(block_number, full_transactions=True)
        return block
    except Exception as e:
        print(f"Error fetching block {block_number}: {e}")
        return None
def format_transaction(tx):
    return {
        'hash': tx.hash.hex(),
        'nonce': tx.nonce,
        'blockHash': tx.blockHash.hex(),
        'blockNumber': tx.blockNumber,
        # ...
        'from': tx['from'],
        'to': tx.to,
        'value': tx.value,
    }
# Formate block data to JSON
def format_block(block):
    block_dict = {
        'number': block.number,
        'hash': block.hash.hex(),
        'nonce': block.nonce.hex(),
        # ...
        'timestamp': block.timestamp,
        'transactions': [format_transaction(tx) for tx in block.transactions],
    }
    return block_dict

def download_block(block_number):
    block = web3.eth.getBlock(block_number, full_transactions=True)
    block_data = {
        'number': block.number,
        'timestamp': block.timestamp,
        'transactions': [tx.hash.hex() for tx in block.transactions]
    }
    with lock:
        with open('blocks.jsonl', 'a') as f:
            f.write(json.dumps(block_data) + '\n')

# download block and write to disk
def download_and_write_blocks(start_block, end_block):
    blocks = []
    with ThreadPoolExecutor(max_workers=20) as executor:
        future_to_block = {executor.submit(get_block, block_number): block_number for block_number in range(start_block, end_block + 1)}
        for future in as_completed(future_to_block):
            block_number = future_to_block[future]
            try:
                block = future.result()
                if block:
                    blocks.append(block)
            except Exception as e:
                print(f"Error in future for block {block_number}: {e}")
    if len(blocks) == 0:
        return
    blocks.sort(key=lambda x: x.number)
    with lock:
        with open("blocks.jsonl", 'a') as f:
            for block in blocks:
                formatted_block = format_block(block)
                f.write(json.dumps(formatted_block) + '\n')

def main(start_block, end_block, batch_size):
    latest_block = web3.eth.block_number
    if start_block > latest_block:
        print("Start block is greater than the latest block on the chain. Exiting.")
        return
    end_block = min(end_block, latest_block)
    for batch_start in range(start_block, end_block + 1, batch_size):
        batch_end = min(batch_start + batch_size - 1, end_block)
        download_and_write_blocks(batch_start, batch_end)

if __name__ == "__main__":
    start_block = 0  # start download block
    end_block = 20000  # end download block
    batch_size = 100  # batch download block size
    main(start_block, end_block, batch_size, json_file)

分析数据并生成图表

使用 Python 读取 JSONL 文件中的交易和区块数据，计算性能指标并生成图表：

import json
import pandas as pd
import plotly.express as px

# 读取 JSONL 文件
def load_jsonl(filename):
    data = []
    with open(filename, 'r') as f:
        for line in f:
            data.append(json.loads(line))
    return data

# 读取交易记录和区块数据
transactions = load_jsonl('transactions.jsonl')
blocks = load_jsonl('blocks.jsonl')

# 将数据转换为 DataFrame
tx_df = pd.DataFrame(transactions)
block_df = pd.DataFrame(blocks)

# 计算每笔交易的执行时间
tx_df['execution_time'] = tx_df['response_time'] - tx_df['start_time']

# 计算每秒发起的交易数
tx_df['start_second'] = (tx_df['start_time'] // 1000) - (tx_df['start_time'].min() // 1000)
tps = tx_df.groupby('start_second').size()

# 计算区块中的交易数
block_tx_count = block_df.explode('transactions').groupby('number').size()

# 生成交易数折线图
fig = px.line(tps, title='每秒发起交易数 (TPS)')
fig.show()

# 生成区块交易数折线图
fig = px.line(block_tx_count, title='每个区块包含的交易数')
fig.show()

通过压力测试和获取收集到的数据，并且生成趋势图的形式，我们可以用图表的方式，可视化的看到在不同的负载情况下，在L2链上进行的交易的趋势变化，有助于理解处理过程中可能出现的瓶颈，从而可以清楚的知道当前部署的Arbitrum链能够处理多少交易，并且根据能力，做出业务决策。

关于测试的思考

通过文章，简单地介绍了，如何使用web3库，使用不同的语言发起交易，下载区块的基本调用流程。

总结这次测试的过程，我觉得最重要的思路是，说是性能评估，其实本质上要做的工作是数据分析，我们只要清晰的定义数据，然后从数据中进行计算，就可以得到相应的指标，计算的过程，就是从原始数据中计算出分组的关键字，然后进行累加的过程。

确定了需要的数据，还有计算指标的过程，剩下的就是从不同的数据源收集数据，为了获得足够数据，有时候可能还要在程序中增加埋点，将需要的信息输出到日志或者数据库中。

整个过程分成几个步骤，首先确定要分析的数据指标、确定之后从数据产生的源头收集数据，收集完数据以后，对数据清洗和转换，然后用转换好的数据，进行分组计算，最后，就可以根据不同的图表和输出的需要，生成结果指标数据集。