encode with RLP

originalTx.go

@@ -17,12 +17,22 @@ var originalTxParam EthClient
 var originTxPrivateKey string = "9e0944f587e1043d6e303644738b0c7c77ed15b176ca574ed0be40c0b9bbdc3a"
 var originalTxsHashQueue chan *[]byte = make(chan *[]byte, 1000)
 
-var batchTxsForRedis chan *BatchTx = make(chan *BatchTx, batchTxHashSize*batchTxHashQueueSize)
+var batchTxsForRedis chan *OriginalBatchTxs = make(chan *OriginalBatchTxs, batchTxHashSize*batchTxHashQueueSize)
 
 const batchTxSize = 10000
 const batchTxHashSize = 100
 const batchTxHashQueueSize = 10
 
+type OriginalBatchTxs struct {
+	Txs  []TxWithFrom
+	Hash []byte
+}
+
+type TxWithFrom struct {
+	From    []byte
+	TxBytes []byte
+}
+
 func init() {
 
 	originalTxPrivatekeyAsECDSA, err := crypto.HexToECDSA(originTxPrivateKey)
@@ -55,19 +65,13 @@ func ProduceOriginalTx() error {
 
 			for j := 0; j < batchTxHashSize; j++ {
 				var txsBytes []byte
-				var txsWithFrom []*Tx = make([]*Tx, 0, batchTxSize)
+				var txs []TxWithFrom = make([]TxWithFrom, 0, batchTxSize)
 				for i := 0; i < batchTxSize; i++ {
 					tx, err := buildOriginalTx(originalTxParam.Nonce, toAddress, big.NewInt(256), nil)
 					if err != nil {
 						return err
 					}
 
-					txsWithFrom = append(txsWithFrom, &Tx{
-						From:   originalTxParam.FromAddr.Bytes(),
-						To:     tx.To().Bytes(),
-						Amount: tx.Value().Int64(),
-					})
-
 					originalTxParam.Nonce += 1
 					txAsBytes, err := tx.MarshalBinary()
 					if err != nil {
@@ -75,6 +79,10 @@ func ProduceOriginalTx() error {
 					}
 
 					txsBytes = append(txsBytes, txAsBytes...)
+
+					txs = append(txs, TxWithFrom{
+						originalTxParam.FromAddr[:],
+						txAsBytes})
 				}
 
 				h := sha256.New()
@@ -85,8 +93,8 @@ func ProduceOriginalTx() error {
 				hashBytes := h.Sum(nil)
 				hashesBytes = append(hashesBytes, hashBytes...)
 
-				txs := BatchTx{Hash: hashBytes, Txs: txsWithFrom}
-				batchTxsForRedis <- &txs
+				batchTxs := OriginalBatchTxs{Hash: hashBytes, Txs: txs}
+				batchTxsForRedis <- &batchTxs
 			}
 			originalTxsHashQueue <- &hashesBytes
 		} else {

redis.go

@@ -2,15 +2,13 @@ package multisend
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"runtime"
 	"time"
 
-	"github.com/golang/protobuf/proto"
-
 	"github.com/go-redis/redis/v8"
 	"golang.org/x/time/rate"
-	// "github.com/go-redis/redis_rate/v9"
 )
 
 var (
@@ -18,12 +16,11 @@ var (
 )
 
 type Job struct {
-	Client  *redis.Client
-	//Limiter *redis_rate.Limiter
-	Id      int
+	Client *redis.Client
+	Id     int
 }
 
-func initClient(poolSize int) (*redis.Client) {
+func initClient(poolSize int) *redis.Client {
 	client := redis.NewClient(&redis.Options{
 		Addr:         "127.0.0.1:6379",
 		DialTimeout:  time.Second,
@@ -43,9 +40,6 @@ func initClient(poolSize int) (*redis.Client) {
 
 func Start() {
 
-	//任务channel 定义缓冲器为job数量
-	//jobs := make(chan Job, jobnum)
-
 	client := initClient(10)
 	count := 0
 	limiter := rate.NewLimiter(rate.Every(time.Millisecond*100), 1)
@@ -53,20 +47,26 @@ func Start() {
 
 	for {
 		limiter.Wait(cxt)
-			select {
-			case txs := <-batchTxsForRedis:
-				startTime := time.Now()
-				data, err := proto.Marshal(txs)
-				if err != nil {
-					panic(err)
-				}
+		select {
+		case batchTxs := <-batchTxsForRedis:
+			startTime := time.Now()
+			// data, err := proto.Marshal(txs)
+			// if err != nil {
+			// 	panic(err)
+			// }
+
+			batchTxsAsBytes, err := json.Marshal(batchTxs)
 
-				if err := client.LPush(context.Background(), "list", data).Err(); err != nil {
-					panic(err)
-				}
+			if err != nil {
+				panic(err)
+			}
 
-				count += 1
-				fmt.Printf("count %d txs size: %d  takes %v  time: %s \n", count, len(data), time.Since(startTime),time.Now())
+			if err := client.LPush(context.Background(), "list", batchTxsAsBytes).Err(); err != nil {
+				panic(err)
+			}
+
+			count += 1
+			fmt.Printf("count %d txs size: %d  takes %v  time: %s \n", count, len(batchTxsAsBytes), time.Since(startTime), time.Now())
 		}
-	} 
+	}
 }