複数のスレッドで発生したデータをとにかく1個のJSONファイルに逐次シリアライズする方法

# 何をしたいのか
こういうことしたい。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/30037/ae47a3a8-c494-47d0-d325-549dc2a6ae80.png)

とりあえずスレッドから飛んできたデータを配列データとして
次から次へとJSONにシリアライズしてファイル保存していきたい。

# メリット
大量のデータを処理するときに逐次保存できるので、メモリ上に巨大なデータ配列を持たなくて良くなる。

# 使うもの
async-std
serde
serde_json
中途半端なRustとマクロの知識


# とりあえずやってみた。

``` Rust
use serde::{
    ser::{SerializeSeq, Serializer},
    Deserialize, Serialize,
};
use serde_json::*;
use std::io::Write;
use std::sync::{mpsc, mpsc::Sender};
use std::sync::{Arc, Mutex};

// JSON化するデータのかたまり
#[derive(Serialize)]
struct ThreadData {
    tid: i64,
    data: u32,
}

// Serialize非同期関数の制御用列挙体
// Dataが来たら、シリアライズして、Finishが来たらスレッドを終了させる
enum SerialDataType {
    Data(ThreadData),
    Finish(i64),
}

// これが、データをシリアライザに適当に送るメソッド。
// データ送信を非同期に行う
// ID（1以上の数値）以下の回数分ループする。そのループ時のインデックスを送る。適当なやつ。
async fn some_heavy_work(id: i64, tx: Sender<SerialDataType>) -> i64 {
    for i in 0..id {
        tx.send(SerialDataType::Data(ThreadData {
            tid: id,
            data: i as u32,
        }));
    }
    tx.send(SerialDataType::Finish(0));
    id
}
// これがシリアライザ。次々来たデータをoutで定義するWriterにとにかく書きまくる
async fn serializer_thread_json(
    threads: usize,
    rx: std::sync::mpsc::Receiver<SerialDataType>,
    out: Arc<Mutex<Box<impl Write>>>,
) {
    let mut threads = threads;
    let mut out = out.lock().unwrap();
    let out = out.as_mut();
    let mut ser = serde_json::Serializer::new(out);
    let mut seq = ser.serialize_seq(None).unwrap();
    for data in rx {
        match data {
            SerialDataType::Data(data) => {
                seq.serialize_element(&data);
            }
            SerialDataType::Finish(_fin) => {
                if threads <= 1 {
                    SerializeSeq::end(seq);
                    return;
                }
                threads -= 1;
            }
        };
    }
    return;
}

#[async_std::main]
async fn main() {
    let worker_max = 10; // データ送信スレッドの個数
    // データ送受信用のチャネル。ここに投げまくるとJSONシリアライズされる
    let (tx, rx) = mpsc::channel();
    let out = Arc::new(Mutex::new(Box::new(std::io::stdout())));
    // チャネルに投げられてきたデータを淡々とjsonにシリアライズするスレッド。
    let ser_th = async_std::task::spawn(serializer_thread_json(worker_max, rx, out));
    let mut worker = vec![];
    // データ送信ワーカを一斉に起動する。
    for i in 0..worker_max {
        worker.push(async_std::task::spawn(some_heavy_work(
            i as i64,
            tx.clone(),
        )));
    }
    // シリアライズが全部完了するまで待つ。
    ser_th.await;
}


```

# おまけ：JSON以外に対応できないのでマクロ化してみた。
マクロ化する以外でいい感じにSerializerを渡すみたいなやつでやろうとしたけど
Serializeトレイトを使ってよしなにしようとしたら、型周りで脳みそ死んだから誰か教えてくだされ

``` Rust
// ここまでは省略
// 飛んできたデータをひたすらシリアライズする
macro_rules! serializer_create {
    ($serializer:expr,$ser_name:ident) => {
        async fn $ser_name(threads: usize, rx: std::sync::mpsc::Receiver<SerialDataType>) {
            let mut threads = threads;
            let mut ser = $serializer;
            let mut seq = ser.serialize_seq(None).unwrap();
            for data in rx {
                match data {
                    SerialDataType::Data(data) => {
                        seq.serialize_element(&data);
                    }
                    SerialDataType::Finish(_fin) => {
                        if threads <= 1 {
                            SerializeSeq::end(seq);
                            return;
                        }
                        threads -= 1;
                    }
                };
            }
            return;
        }
    };
}

#[async_std::main]
async fn main() {
    serializer_create!(
        serde_json::Serializer::new(std::io::stdout()),
        serializer_serde_json
    );

    let worker_max = 10;
    let (tx, rx) = mpsc::channel();
    let out = Arc::new(Mutex::new(Box::new(std::io::stdout())));
    let ser_th = async_std::task::spawn(serializer_serde_json(worker_max, rx));
    let mut worker = vec![];
    for i in 0..worker_max {
        worker.push(async_std::task::spawn(some_heavy_work(
            i as i64,
            tx.clone(),
        )));
    }
    ser_th.await;
}
```