如何通过DynamoDB全局二级索引批量查询指定邮箱账户？

DynamoDB的全局二级索引（GSI）主键不支持IN条件表达式，无法通过单个Query请求直接查询多个email对应的记录。你需要对每个目标email单独发起Query请求，再合并结果。用Go的goroutine并行执行这些请求，可提升处理效率。

具体实现步骤 #

1. 并行发起多Query请求 #

针对每个目标email，创建独立的Query请求，指定GSI作为查询索引，用EQ条件匹配email字段。通过goroutine并行执行所有请求，等待全部完成后汇总结果。

2. 代码示例 #

package main

import (
	"context"
	"fmt"
	"sync"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

type Account struct {
	AccountID string
	Email     string
	FirstName string
	LastName  string
}

// QueryAccountsByEmails 批量查询指定邮箱的账户
func QueryAccountsByEmails(ctx context.Context, client *dynamodb.Client, tableName, indexName string, emails []string) ([]Account, error) {
	var wg sync.WaitGroup
	results := make([]Account, 0, len(emails))
	errChan := make(chan error, len(emails))
	resultChan := make(chan Account, len(emails))

	for _, email := range emails {
		wg.Add(1)
		go func(e string) {
			defer wg.Done()

			input := &dynamodb.QueryInput{
				TableName:              aws.String(tableName),
				IndexName:              aws.String(indexName), // 替换为你的GSI名称
				KeyConditionExpression: aws.String("email = :e"),
				ExpressionAttributeValues: map[string]types.AttributeValue{
					":e": &types.AttributeValueMemberS{Value: e},
				},
				ProjectionExpression: aws.String("account_id, email, first_name, last_name"), // 指定需要返回的字段
			}

			resp, err := client.Query(ctx, input)
			if err != nil {
				errChan <- fmt.Errorf("查询邮箱%s失败: %w", e, err)
				return
			}

			// 解析返回结果（假设每个email对应唯一记录）
			for _, item := range resp.Items {
				account := Account{
					AccountID: item["account_id"].(*types.AttributeValueMemberS).Value,
					Email:     item["email"].(*types.AttributeValueMemberS).Value,
					FirstName: item["first_name"].(*types.AttributeValueMemberS).Value,
					LastName:  item["last_name"].(*types.AttributeValueMemberS).Value,
				}
				resultChan <- account
			}
		}(email)
	}

	// 等待所有goroutine完成后关闭通道
	go func() {
		wg.Wait()
		close(resultChan)
		close(errChan)
	}()

	// 收集查询结果
	for res := range resultChan {
		results = append(results, res)
	}

	// 检查是否存在错误
	if len(errChan) > 0 {
		return results, <-errChan // 返回第一个错误，可根据需求修改为收集所有错误
	}

	return results, nil
}

func main() {
	// 初始化DynamoDB客户端（示例省略，需根据你的配置实现）
	// cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-1"))
	// if err != nil {
	// 	log.Fatalf("加载配置失败: %v", err)
	// }
	// client := dynamodb.NewFromConfig(cfg)

	targetEmails := []string{"linda@gmail.com", "harry@gmail.com"}
	accounts, err := QueryAccountsByEmails(context.Background(), nil, "your-table-name", "email-index", targetEmails)
	if err != nil {
		fmt.Printf("查询失败: %v\n", err)
		return
	}

	for _, acc := range accounts {
		fmt.Printf("账户信息: %+v\n", acc)
	}
}

注意事项 #

• 并发控制：如果目标email数量较多，建议用带缓冲的通道限制并发请求数，避免触发DynamoDB限流。
• 分页处理：若业务允许同一个email对应多条记录，需处理LastEvaluatedKey，循环查询直到获取全部结果。
• 错误处理：示例仅返回第一个错误，可根据业务需求修改为收集所有错误，或跳过查询失败的邮箱继续处理其他请求。

内容的提问来源于stack exchange，提问作者ChrisS