利用 gRPC 的实时数据——一个详细的 Pump.fun 指南

访问实时链上数据对开发者、研究人员和交易机器人至关重要。虽然有多种方法可以流式传输这些数据，gRPC 是最快和效率最高的，提供低延迟和更简单的代码。

Pump.fun 是一个受欢迎的 Solana 平台，用户可以在此启动价格低于 2 美元的可即时交易的币种。随着数以百万计的 meme 代币被推出，每分钟还有成千上万个新币诞生，开发者需要实时更新以保持领先。gRPC 以毫秒为单位提供这些更新。在本文中，您将学习如何使用 gRPC 实时流式传输 Pump.fun 事件。

开始之前

要开始，我们需要一些东西。

身份验证：您的 Shyft API 密钥、gRPC 端点和 gRPC 令牌

您可以从 Shyft 网站 获取自己的 Shyft API 密钥（Shyft 使用的身份验证参数）。您还可以在 Shyft 控制面板 上找到您的区域特定 gRPC 端点和访问令牌。

一个服务器端后端（例如 NodeJS）以接收 gRPC 数据

由于 gRPC 服务在网页浏览器中不受支持，您需要一个后端应用程序来接收 gRPC 数据。在这个例子中，我们使用了 NodeJS，但任何其他 服务器端环境，例如 C#，Go，Java，Kotlin，Python 或 PHP 也可以使用。

介绍 — 要涵盖的主题

在本文中，我们将研究如何使用 gRPC 流式传输 Pump.fun 上的交易交易。我们将涵盖以下主题，

• 克隆和安装依赖项
• 设置与 Solana 网络的 gRPC 连接
• 订阅 Pump.fun 上交易交易的流式端点
• 实时处理和反序列化流式数据

克隆和安装

要开始，我们克隆以下内容并安装依赖项：

$ git clone https://github.com/Shyft-to/solana-defi.git

cd grpc-pump
npm install

与本文相关的所有代码可以在我们的 GitHub 这里 找到。请随意克隆并跟随。

设置与 Solana 网络的 gRPC 连接

我们使用 gRPC 从 Pump.fun 流式传输数据的第一步是设置我们的客户端服务。它需要您的 gRPC URL 和 X_TOKEN，格式如下：

const client = new Client(
    '你的区域特定 Shyft gRPC URL',
    'Shyft gRPC 访问令牌',
    undefined,
  );

现在我们可以继续从区块链获取数据。

订阅 Pump.fun 上交易交易的流式端点

我们的目标是使用 gRPC 从 Solana 区块链流式传输特定数据。为此，我们将使用 handleStream() 函数，该函数需要两个参数：

1. client: Client：此参数用于订阅链上事件并与 Solana 区块链建立连接。
2. arg: SubscriberRequest：此参数指定我们要从区块链获取的数据。

通过这种方式使用 handleStream() 函数，我们可以有效且方便地流式传输 Solana 区块链上的实时链上数据。

interface SubscribeRequest {
    accounts: { [key: string]: SubscribeRequestFilterAccounts };
    slots: { [key: string]: SubscribeRequestFilterSlots };
    transactions: { [key: string]: SubscribeRequestFilterTransactions };
    transactionsStatus: { [key: string]: SubscribeRequestFilterTransactions };
    blocks: { [key: string]: SubscribeRequestFilterBlocks };
    blocksMeta: { [key: string]: SubscribeRequestFilterBlocksMeta };
    entry: { [key: string]: SubscribeRequestFilterEntry };
    commitment?: CommitmentLevel | undefined;
    accountsDataSlice: SubscribeRequestAccountsDataSlice[];
    ping?: SubscribeRequestPing | undefined;
  }

async function handleStream(client: Client, args: SubscribeRequest) {
    // 订阅事件
    const stream = await client.subscribe();

    // 创建 `error` / `end` 处理程序
    const streamClosed = new Promise<void>((resolve, reject) => {
      stream.on("error", (error) => {
        console.log("错误", error);
        reject(error);
        stream.end();
      });
      stream.on("end", () => {
        resolve();
      });
      stream.on("close", () => {
        resolve();
      });
    });

    // 处理更新
    stream.on("data", async (data) => {
      try{
        console.log(data);
      }catch(error){
        if(error){
          console.log(error)
        }
      }
    });

    // 发送订阅请求
    await new Promise<void>((resolve, reject) => {
      stream.write(args, (err: any) => {
        if (err === null || err === undefined) {
          resolve();
        } else {
          reject(err);
        }
      });
    }).catch((reason) => {
      console.error(reason);
      throw reason;
    });

    await streamClosed;
  }

在为 handleStream() 函数定义 SubscriberRequest 参数后，下一步是指定我们要从 Solana 区块链获取的数据。在本教程中，我们专注于获取 pump.fun 上的代币交易。为此，我们需要定义一个 SubscriberRequest 对象，指定我们的数据请求参数。此对象应包含以下信息：

• 我们要获取的数据类型（在此情况下为代币交易）
• pump.fun 的地址
• 限制数据获取的任何额外过滤器或参数

以下是 fetching pump.fun 上的代币交易的 SubscriberRequest 对象的示例：

const req = {
    accounts: {},
    slots: {},
    transactions: {
      bondingCurve: {
        vote: false,
        failed: false,
        signature: undefined,
        accountInclude: ['6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P'], //地址 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P
        accountExclude: [],
        accountRequired: [],
      },
    },
    transactionsStatus: {},
    entry: {},
    blocks: {},
    blocksMeta: {},
    accountsDataSlice: [],
    ping: undefined,
    commitment: CommitmentLevel.CONFIRMED, //用于接收确认的交易更新
  };

地址 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P 代表在 Solana 区块链上 "pump.fun" 平台的链上地址。然而，仅凭这个地址不足以获取所需的数据。为了访问实时链上数据，我们需要定义 SubscribeCommand 函数。此函数将用于指定我们的数据请求参数并订阅 Solana 区块链上的流式端点。SubscribeCommand 函数应将 SubscriberRequest 对象作为参数，并用它来与 Solana 区块链建立连接并订阅所需的流式端点。

async function subscribeCommand(client: Client, args: SubscribeRequest) {
    while (true) {
      try {
        await handleStream(client, args);
      } catch (error) {
        console.error("流式传输错误，1秒后重启...", error);
        await new Promise((resolve) => setTimeout(resolve, 1000));
      }
    }
  }

并调用它

subscribeCommand(client, req);

我们的输出应如下所示

{
  filters: [ 'bondingCurve' ],
  account: undefined,
  slot: undefined,
  transaction: {
    transaction: {
      signature: <Buffer 18 f2 41 ca 0d c3 2d 24 a3 4c 25 0b 0f d5 e0 bd fd 75 35 a1 ed a3 43 8a f8 fc 86 c0 01 08 70 32 f9 ac 53 3c 46 e8 e2 62 22 4c 79 ab 3f f5 4b d5 81 0c ... 14 more bytes>,
      isVote: false,
      transaction: [Object],
      meta: [Object],
      index: '315'
    },
    slot: '283417510'
  },
  block: undefined,
  ping: undefined,
  pong: undefined,
  blockMeta: undefined,
  entry: undefined
}

尽管这是我们请求的数据，但这并不是我们希望获得的数据。我们已经接近了，我们只需对输出进行反序列化以使其可理解。

实时处理和反序列化流式数据

为了将来自流式端点的接收数据转换为更易读和可用的格式，需要一个被称为反序列化的过程。以下是您可以编写的反序列化数据的函数示例：

import base58 from "bs58";
export function decodeTransact(data){
    const output = base58.encode(Buffer.from(data,'base64'))
    return output;
}

该函数将任何缓冲数据解码为更可用的格式。

接下来，我们希望以更格式化的方式定义我们的输出。

export function tOutPut(data){
    const dataTx = data.transaction.transaction
    const signature = decodeTransact(dataTx.signature);
    const message = dataTx.transaction?.message
    const header = message.header;
    const accountKeys = message.accountKeys.map((t)=>{
        return  decodeTransact(t)
    })
    const recentBlockhash =  decodeTransact(message.recentBlockhash);
    const instructions = message.instructions
    const meta = dataTx?.meta
    return {
        signature,
        message:{
           header,
           accountKeys,
           recentBlockhash,
           instructions
        },
        meta
    }
}

然后我们重写我们的 handleStream() 函数以适应更新后的代码。

async function handleStream(client: Client, args: SubscribeRequest) {
    // 订阅事件
    const stream = await client.subscribe();

    // 创建 `error` / `end` 处理程序
    const streamClosed = new Promise<void>((resolve, reject) => {
      stream.on("error", (error) => {
        console.log("错误", error);
        reject(error);
        stream.end();
      });
      stream.on("end", () => {
        resolve();
      });
      stream.on("close", () => {
        resolve();
      });
    });

    // 处理更新
    stream.on("data", async (data) => {
      try{
        const result = await tOutPut(data);
        console.log(result);
      }catch(error){
        if(error){
          console.log(error)
        }
      }
    });

    // 发送订阅请求
    await new Promise<void>((resolve, reject) => {
      stream.write(args, (err: any) => {
        if (err === null || err === undefined) {
          resolve();
        } else {
          reject(err);
        }
      });
    }).catch((reason) => {
      console.error(reason);
      throw reason;
    });

    await streamClosed;
  }

最后，我们用 SubscribeCommand 函数进行调用。

subscribeCommand(client, req);

我们的更新后的输出应如下所示：

signature: '4aTaVifyJNwaxrqP1939p6Cwi7doEJ627eo7XV1dC6cHHDLVf5JJ8qSdsVzQbmFRd2vWDK7sDZRHWWvuWwsnaCKj',
  message: {
    header: {
      numRequiredSignatures: 1,
      numReadonlySignedAccounts: 0,
      numReadonlyUnsignedAccounts: 8
    },
    accountKeys: [\
      'Dn2sbLk6cMYyRu9D5h1G66vUGFTE7BRgqho9X6c52wLE',\
      'CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM',\
      'Gf72ajaDxUPSgXbRCnYjAXpPQXofam6FmedepfYJNY4B',\
      '7GXjNDdNkSnsmaXmWUy4Woh6NAQgMCABBEPWtHyLBJ9J',\
      'CcCAgQcArc6jjF2zAaDUzQ1FKgisqDjubCNJvVmPHVMU',\
      'ZG98FUCjb8mJ824Gbs6RsgVmr1FhXb2oNiJHa2dwmPd',\
      'ComputeBudget111111111111111111111111111111',\
      '6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P',\
      '4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf',\
      'C1Fabt6rjQGcS68obCEH6zuywWfzpfvCCfKBJntCpump',\
      '11111111111111111111111111111111',\
      'ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL',\
      'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA',\
      'Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1'\
    ],
    recentBlockhash: 'EGc4SgYd4xmex8ZS4LQcjtvzBW8aU3yafphxscM9hBTa',
    instructions: [ [Object], [Object], [Object] ]
  },
  meta: {
    err: undefined,
    fee: '405000',
    preBalances: [\
      '822302413',  '272648835990597',\
      '1191944363', '2039280',\
      '2039280',    '13219204698',\
      '1',          '1141440',\
      '2500000',    '1461600',\
      '1',          '731913600',\
      '934087680',  '0'\
    ],
    postBalances: [\
      '843092001',  '272648836206849',\
      '1170319136', '2039280',\
      '2039280',    '13219419085',\
      '1',          '1141440',\
      '2500000',    '1461600',\
      '1',          '731913600',\
      '934087680',  '0'\
    ],
    innerInstructions: [ [Object] ],
    innerInstructionsNone: false,
    logMessages: [\
      '程序 ComputeBudget111111111111111111111111111111 invoke [1]',\
      '程序 ComputeBudget111111111111111111111111111111 success',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P invoke [1]',\
      '程序日志：指令: 卖出',\
      '程序 TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [2]',\
      '程序日志：指令: 转移',\
      '程序 TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA 使用 4645 的 381291 计算单位',\
      '程序 TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P invoke [2]',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P 使用 2003 的 373161 计算单位',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P success',\
      '程序数据: vdt/007mYe6jgQqFQndU+0Owx6KwKFXj5LXaRYNPQ2h7er5FKWGLH4v5SQEAAAAAL/LStqYAAAAAvdV/G53p8Qe4ABBpTvcEEXUBRn8h6KfMWZcLzg+Qi3tazrtmAAAAAPCG0kEHAAAAuu/G0EirAwDw2q5FAAAAALpXtIS3rAIA',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P 使用 30418 的 399850 计算单位',\
      '程序 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P success',\
      '程序 11111111111111111111111111111111 invoke [1]',\
      '程序 11111111111111111111111111111111 success'\
    ],
    logMessagesNone: false,
    preTokenBalances: [ [Object], [Object] ],
    postTokenBalances: [ [Object], [Object] ],
    rewards: [],
    loadedWritableAddresses: [],
    loadedReadonlyAddresses: [],
    returnData: undefined,
    returnDataNone: true,
    computeUnitsConsumed: '30718'
  }
}

结论

使用专用节点在 Solana 区块链上流式传输数据确实是一项复杂且资源密集的任务，需要更大和更复杂的代码库。然而，使用 gRPC，开发者现在可以方便地流式传输链上数据，简化代码，专注于构建强大的应用程序，利用 Solana 区块链的潜力。在 Shyft 上，我们很高兴向开发者社区介绍这一创新解决方案。我们的团队致力于提供必要的工具和支持，使这一过程顺利进行。我们鼓励开发者加入我们的 Discord 服务器或在 Twitter 上关注我们以获取更新和进一步的帮助。使用 gRPC，开发者可以利用低延迟和减少的代码库轻松构建复杂的项目。

与本文相关的所有代码可以在我们的 [GitHub 这里](https://github.com/Shyft-to/solana-defi/tree/main/grpc-pump) 找到。请随意克隆并跟随。要了解更多关于 Solana 的数据流式传输，请查看我们关于 [在 Solana 上流式实时数据](https://blogs.shyft.to/how-to-stream-real-time-solana-transactions-using-shyfts-grpc-service-1a8f8ad44da3) 或 [追踪 Raydium 上的新池](https://blogs.shyft.to/how-to-track-new-pools-on-raydium-with-shyft-grpc-2497df832ea0) 的其他文章。

资源

• SHYFT API 文档
• Shyft 网站
• 获取 API 密钥
• GitHub
• Javascript SDK
• 在 Swagger UI 上尝试我们的 API

• 原文链接： blogs.shyft.to/how-to-st...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，在这里修改，还请包涵～