问题[multithreading](coding)

我有一个自制的迭代器，我想循环它，在线程池中执行一些昂贵的处理，并按输入顺序收集结果。

我使用这个链：
Iterator > enumerate() > rayon par_bridge() > map()

当我使用此链并将collect()其放入向量中时，没有任何问题，但当我使用 for 循环时，我收到错误，提示某些内容不是迭代器。
我不明白其中的区别。

这是 MRE：

use std::vec::IntoIter;
use rayon::prelude::*;

fn main() {
    // collecting();
    for_loop();
}

fn collecting() {
    let some_vec = (1..100).collect::<Vec<u8>>();
    let x_iter = AA {y: some_vec.into_iter()};
    
    let with_index: Vec<(usize, u8)> = x_iter
        .enumerate()
        .par_bridge()
        .map(|(i, x)| {(i, x - 1)})
        .collect();

    println!("{:?}", with_index);
}

fn for_loop() {
    let some_vec = (1..100).collect::<Vec<u8>>();
    let x_iter = AA {y: some_vec.into_iter()};

    let mut with_index: Vec<(usize, u8)> = Vec::with_capacity(99);
    for x in x_iter
        .enumerate()
        .par_bridge() // with this line commented out it works fine
        .map(|(i, x)| {(i, x - 1)})
    {
        with_index.push(x);
    }
    println!("{:?}", with_index);
}


struct AA {
    y: IntoIter<u8>
}

impl Iterator for AA {
    type Item = u8;

    fn next(&mut self) -> Option<u8> {
        self.y.next()
    }
}

该collecting函数运行正常。
该for_loop函数给出以下错误：

rayon::iter::Map<IterBridgestd::iter::Enumerate<AA>, {closure@src/main.rs:30:14: 30:22}> 不是迭代器

请解释我为什么会收到这个错误。

我正在尝试在 Delphi 10.4 (Sydney) 中实现多线程。我使用 TTask 并发运行任务，使用 TThreadPool 限制同时运行的线程。

这是我的代码：

var
  MyThreadPool: TThreadPool;
  I: Integer;
  Tasks: TArray<ITask>;

function ProcessTask(const TaskId: Integer): TProc;
begin
  Result := procedure
  Begin
    WriteLn(Format('Started TaskId: %d ThreadId: %d',[TaskId,TThread.Current.ThreadID]));
    Sleep(2000); // Simulate work
    WriteLn(format('TaskId %d completed. ThreadId: %d',[TaskId, TThread.Current.ThreadID]));
  End;
end;

begin
  try

    Writeln('Creating TThreadPool');
    MyThreadPool := TThreadPool.Create;

    try
      MyThreadPool.SetMinWorkerThreads(1);
      MyThreadPool.SetMaxWorkerThreads(4);

      // Hold 10 tasks
      SetLength(Tasks,10);

      for I := 0 to High(Tasks) do
      Begin
        Tasks[I] := TTask.Create(
          ProcessTask(I+1),
          MyThreadPool
        );

        // Start the task
        Tasks[I].Start;
      End;

      // Wait for all tasks to complete
      for I := 0 to High(Tasks) do
        Tasks[I].Wait;

      WriteLn('All tasks completed.');
      ReadLn;

    finally
      MyThreadPool.Free;
    end;

  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
end.

上述代码的输出是：

Creating TThreadPool
Started TaskId: 1 ThreadId: 17292
Started TaskId: 1 ThreadId: 17292
Started TaskId: 2 ThreadId: 19256
Started TaskId: 3 ThreadId: 25268
Started TaskId: 3 ThreadId: 25268
Started TaskId: 4 ThreadId: 21112
TaskId 1 completed. ThreadId: 17292
Started TaskId: 5 ThreadId: 17292
TaskId 2 completed. ThreadId: 19256
Started TaskId: 6 ThreadId: 19256
TaskId 3 completed. ThreadId: 25268
TaskId 3 completed. ThreadId: 25268
TaskId 4 completed. ThreadId: 21112
Started TaskId: 7 ThreadId: 21112Started TaskId: 8 ThreadId: 25268Started TaskId: 7 ThreadId: 21112Started TaskId: 8 ThreadId: 25268
TaskId 5 completed. ThreadId: 17292
Started TaskId: 9 ThreadId: 17292
TaskId 6 completed. ThreadId: 19256
Started TaskId: 10 ThreadId: 19256
TaskId 8 completed. ThreadId: 25268
TaskId 8 completed. ThreadId: 25268
TaskId 7 completed. ThreadId: 21112
TaskId 9 completed. ThreadId: 17292
TaskId 10 completed. ThreadId: 19256
All tasks completed.

代码的问题在于在某些情况下开始和完成报告会重复：

1. Started TaskId出现了两次，分别为 TaskId 1,3,7,8。
2. TaskId完成出现了两次，分别是 TaskId 3,8。

我已经使用 Delphi 有一段时间了，现在正尝试在我的代码中实现多线程，要求如下：

1. 每次只能运行 2 个线程。

2. 其他任务应在队列中等待，并在线程可用时执行。

我尝试使用TParallel.For来实现这一点，但无法实现所需的行为。这是我的示例代码：

    type
       TMyRecord = record
         msg: String;
         sleep: Integer;
          end;

    var
      Form2: TForm2;

    implementation

    {$R *.dfm}

    procedure TForm2.Button1Click(Sender: TObject);
    var
      record1, record2, record3, record4, record5: TMyRecord;
      MyList: TList<TMyRecord>;
    begin
      // Initialize records
      record1.msg := 'Item1';
      record1.sleep := 10;

      record2.msg := 'Item2';
      record2.sleep := 10;

      record3.msg := 'Item3';
      record3.sleep := 3000;

      record4.msg := 'Item4';
      record4.sleep := 3000;

      record5.msg := 'Item5';
      record5.sleep := 100;

     MyList := TList<TMyRecord>.Create;
      try
        // Add records to the list
        MyList.Add(record1);
        MyList.Add(record2);
        MyList.Add(record3);
        MyList.Add(record4);
        MyList.Add(record5);

    // Use TParallel.For
    TParallel.For(0, MyList.Count - 1,
      procedure (i: Integer)
      var
        ListItem: TMyRecord;
      begin
        ListItem := MyList[i];

        TThread.Synchronize(nil,
          procedure
          begin
            Memo1.Lines.Add(ListItem.msg);
          end);

        Sleep(ListItem.sleep);
      end);

      finally
    MyList.Free;
  end;
end;

我也尝试使用TTask.Run和ThreadPool，但无法实现所需的行为。更新 - 这是我尝试使用 Threadpool 执行的操作

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
  ThreadPool: TThreadPool;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    // Add records to the list
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    ThreadPool := TThreadPool.Create;
    ThreadPool.SetMaxWorkerThreads(2);

    // Iterate through the list and queue each work item with a copy of the record
    for var i := 0 to MyList.Count - 1 do
    begin
      var ListItem := MyList[i];

      ThreadPool.QueueWorkItem(
        procedure
        var
          LocalItem: TMyRecord;
        begin
          LocalItem := ListItem;

          TThread.Queue(nil,
            procedure
            begin
              Memo1.Lines.Add(LocalItem.msg);
              Sleep(LocalItem.sleep);  // Simulate the delay
            end);
        end);
    end;
  finally
    MyList.Free;
  end;
end;

此代码有问题-

1. 它仅打印“item5”5次。
2. 等待时间（睡眠）不起作用。

我该如何修改此代码以将同时运行的线程数限制为 2？并在线程可用时将其他任务排队执行？

我正在尝试使用布尔标志来表示线程应该退出。但是，目前我的代码违反了借用检查器规则。我理解它为什么违反这些规则，但我不知道最合适的设计是什么以及如何解决问题。

fn long_running(exit_flag: &bool) {

    loop {
        std::thread::sleep(std::time::Duration::from_secs(10));

        if *exit_flag {
            break;
        }
    }
}

fn main() {

    let mut exit_flag = false;
    std::thread::scope(
        |scope| {
            let handle = scope.spawn(
                || {
                    long_running(&exit_flag); # (1)
                }
            );

            exit_flag = true; # (2)

            // terminate the consumer poll loop
            handle.join().expect("failed to join thread");
        }
    );
}

问题涉及标有# (1)和 的行# (2)。

• # (1)：引用exit_flag自
• # (2)：在引用仍被借用时尝试改变底层变量

由于借用检查规则，这是不允许的。

• 解决该问题的一个方法可能是堆分配标志，并使用它在两个线程之间共享数据。
• 但是，使用的目的std::thread::scope是允许启动一个可以从父（主）线程的堆栈访问局部变量的线程。

顺便说一下，这个想法是从这个帖子里得到的

• 如何将堆栈变量的引用传递给线程？

如果我在 Raylib 中不使用线程加载纹理，它可以正常工作：

INFO: FILEIO: [assets/spades/5_spades.png] File loaded successfully
INFO: IMAGE: Data loaded successfully (112x176 | R8G8B8 | 1 mipmaps)
INFO: TEXTURE: [ID 46] Texture loaded successfully (112x176 | R8G8B8 | 1 mipmaps)

如果我调用该函数在线程内加载纹理，则会失败：

INFO: FILEIO: [assets/spades/7_spades.png] File loaded successfully
INFO: IMAGE: Data loaded successfully (112x176 | R8G8B8 | 1 mipmaps)
WARNING: TEXTURE: Failed to load texture

但它没有说明为什么加载纹理失败。

我已在此处发布了源代码： https： //github.com/imekon/RaylibThreading

我在平台上看到了很多关于使用线程更新 ProgressBar 的代码，但所有这些都与执行某些计算或更新某些控件属性有关。我不知道这是否可行，但我需要的是 ProgressBar 显示创建 225 (15 X 15) 到 2500 (50 x 50) 个 TextField 的进度。以下是文件和 .fxml。当我包含线程代码部分时，网格停止出现。提前谢谢您。

应用程序

网格应用程序.java

import javafx.application.Application;
import javafx.fxml.FXMLLoader;
import javafx.scene.Scene;
import javafx.stage.Stage;

import java.io.IOException;

public class GridCenterApplication extends Application {
    @Override
    public void start(Stage stage) throws IOException {
        FXMLLoader fxmlLoader = new FXMLLoader(GridCenterApplication.class.getResource("main-view.fxml"));
        Scene scene = new Scene(fxmlLoader.load());
        stage.setTitle("Grid");
        stage.setScene(scene);
        stage.setMaximized(true);
        stage.show();
    }

    public static void main(String[] args) {
        launch();
    }
}

网格控制器.java

import javafx.event.ActionEvent;
import javafx.fxml.FXML;
import javafx.fxml.Initializable;
import javafx.geometry.Pos;
import javafx.scene.control.*;
import javafx.scene.layout.GridPane;

import java.net.URL;
import java.util.ResourceBundle;

public class GridCenterController implements Initializable {

    @FXML
    private ScrollPane scpGrid;

    @FXML
    private Spinner<Integer> spnCols;

    @FXML
    private Spinner<Integer> spnRows;

    @FXML
    private ProgressBar pgbProgress;

    GridPane gridPane;

    private final int CELL_HORIZONTAL_GAP = 1;
    private final int CELL_VERTICAL_GAP = 1;
    private final int CELL_HORIZONTAL_SIZE = 40;
    private final int CELL_VERTICAL_SIZE = 40;

    private int totalCols = 0;
    private int totalRows = 0;

    @FXML
    void onMnuItemNewGridAction(ActionEvent event) {

        if(!(scpGrid.getContent() == null)){
            scpGrid.setContent(null);
        }

        totalCols = spnCols.getValue();
        totalRows = spnRows.getValue();

        var newGrid = new Grid(totalCols, totalRows, CELL_HORIZONTAL_GAP, CELL_VERTICAL_GAP, CELL_HORIZONTAL_SIZE,
                CELL_VERTICAL_SIZE, pgbProgress);
        gridPane = newGrid.getGrid();
        scpGrid.setContent(gridPane);
    }

    @Override
    public void initialize(URL url, ResourceBundle resourceBundle) {
        SpinnerValueFactory<Integer> numberOfCols = new SpinnerValueFactory.IntegerSpinnerValueFactory(15, 50);
        SpinnerValueFactory<Integer> numberOfRows = new SpinnerValueFactory.IntegerSpinnerValueFactory(15, 50);
        spnCols.setValueFactory(numberOfCols);
        spnRows.setValueFactory(numberOfRows);
        scpGrid.contentProperty().addListener((observableValue, oldValue, newValue) -> {
            if (newValue != null && newValue.isVisible()) {
               pgbProgress.setProgress(0);
            }
        });

    }
}

网格.java

import javafx.concurrent.Task;
import javafx.scene.control.ProgressBar;
import javafx.scene.control.TextField;
import javafx.scene.layout.GridPane;

public class Grid {

    private final GridPane grid;

    public Grid(int totalCols, int totalRows, int CELL_HORIZONTAL_GAP, int CELL_VERTICAL_GAP, int CELL_HORIZONTAL_SIZE,
                int CELL_VERTICAL_SIZE, ProgressBar pgbProgress) {

        grid = new GridPane();
        grid.setHgap(CELL_HORIZONTAL_GAP);
        grid.setVgap(CELL_VERTICAL_GAP);
        TextField[][] arrayLetterField = new TextField[totalCols][totalRows];

        Task<Void> task = new Task<Void>() {
            @Override
            protected Void call() throws Exception {

                pgbProgress.setProgress(0);
                double total = totalCols * totalRows;
                double i = 1.0;

                for (int row = 0; row < totalRows; row++) {
                    for (int col = 0; col < totalCols; col++) {
                        arrayLetterField[col][row] = new TextField();
                        arrayLetterField[col][row].setMinSize(CELL_HORIZONTAL_SIZE, CELL_VERTICAL_SIZE);
                        arrayLetterField[col][row].setMaxSize(CELL_HORIZONTAL_SIZE, CELL_VERTICAL_SIZE );
                        grid.add(arrayLetterField[col][row], col, row);
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }

                        updateProgress(i, total);
                        i++;
                    }
                }
                return null;
            }
        };

        pgbProgress.progressProperty().bind(task.progressProperty());
        new Thread(task).start();

    }

    public GridPane getGrid() {
        return grid;
    }

}

主视图.fxml

<?xml version="1.0" encoding="UTF-8"?>

<?import javafx.scene.control.Label?>
<?import javafx.scene.control.Menu?>
<?import javafx.scene.control.MenuBar?>
<?import javafx.scene.control.MenuItem?>
<?import javafx.scene.control.ProgressBar?>
<?import javafx.scene.control.ScrollPane?>
<?import javafx.scene.control.Spinner?>
<?import javafx.scene.layout.BorderPane?>
<?import javafx.scene.layout.Pane?>
<?import javafx.scene.layout.VBox?>

<BorderPane prefHeight="767.0" prefWidth="1053.0" xmlns="http://javafx.com/javafx/17" xmlns:fx="http://javafx.com/fxml/1" fx:controller="com.example.gridpanetest.GridCenterController">
    <top>
        <VBox prefWidth="100.0" BorderPane.alignment="CENTER">
            <children>
                <MenuBar fx:id="mnuBar" prefHeight="25.0" prefWidth="360.0">
                    <menus>
                        <Menu mnemonicParsing="false" text="Grid">
                            <items>
                                <MenuItem mnemonicParsing="false" onAction="#onMnuItemNewGridAction" text="New grid" />
                            </items>
                        </Menu>
                    </menus>
                </MenuBar>
            <Pane prefHeight="80.0" prefWidth="1053.0">
               <children>
                  <Label layoutX="26.0" layoutY="15.0" text="Columns" />
                  <Label layoutX="26.0" layoutY="46.0" text="Rows" />
                  <Spinner fx:id="spnCols" layoutX="79.0" layoutY="11.0" prefHeight="25.0" prefWidth="57.0" />
                  <Spinner fx:id="spnRows" layoutX="79.0" layoutY="42.0" prefHeight="25.0" prefWidth="57.0" />
                  <ProgressBar fx:id="pgbProgress" focusTraversable="false" layoutX="185.0" layoutY="31.0" prefWidth="200.0" progress="0.0" />
               </children>
            </Pane>
            </children>
        </VBox>
    </top>
   <center>
        <ScrollPane fx:id="scpGrid" style="-fx-background-color: #dbbb92; -fx-background: #dbbb92;" BorderPane.alignment="CENTER" />
   </center>
</BorderPane>

我使用线程池函数构建管道，并将 context.Context 作为参数传入其中。当调用 cancel() 函数或超时时，管道必须正常终止，这样就不会剩下任何可用的 goroutine。

我使用的职能：

func generate(amount int) <-chan int {
    result := make(chan int)
    go func() {
        defer close(result)
        for i := 0; i < amount; i++ {
            result <- i
        }
    }()

    return result
}

func sum(input <-chan int) int {
    result := 0
    for el := range input {
        result += el
    }
    return result
}

func process[T any, R any](ctx context.Context, workers int, input <-chan T, do func(T) R) <-chan R {
    wg := new(sync.WaitGroup)
    result := make(chan R)

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-input:
                    if !ok {
                        return
                    }
                    select {
                    case <-ctx.Done():
                        return
                    case result <- do(val):
                    }
                }
            }
        }()
    }

    go func() {
        defer close(result)
        wg.Wait()
    }()
    return result
}

用法：

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 1200*time.Millisecond)
    defer cancel()

    input := generate(1000)
    multiplied := process(ctx, 15, input, func(val int) int {
        time.Sleep(time.Second)
        return val * 2
    })
    increased := process(ctx, 15, multiplied, func(val int) int {
        return val + 10
    })

    fmt.Println("Result: ", sum(increased)) //  360 is ok
    fmt.Println("Num goroutine: ", runtime.NumGoroutine())  // 18 is too much
}

我理解发生这种情况是因为所有增加 goroutine 都结束了，而乘法 goroutine 仍在运行。

有没有什么规范的方法来解决这个问题？

我正在考虑将拥有一些线程安全值的闭包传递给生成的线程的可能性。然后该线程将能够仅通过知道签名来调用某些东西，而其内容对它来说是不透明的。

我写了一个简单的测试，令人惊讶的是，它成功了。为了使闭包可以在线程内调用，Rust 编译器建议为其添加std::marker::Send和'static约束。第一个适用是有道理的——闭包捕获的所有值都是Send，但静态生命周期要求被认为已满足，这让我很困惑，因为我假设编译器会认为闭包及其拥有的值具有特定的非静态生命周期。

我想了解原因。代码如下：

use std::sync::{mpsc::{Receiver, Sender}, Arc, Mutex};

fn main() {
    let tx = start_thread();

    // can also run in a loop with a sleep in between or whatever
    let _ = tx.send(String::from("Something"));
}

fn start_thread() -> Sender<String> {
    // Something created on the main thread that we pass to the spawned thread
    //  inside the closure -- the spawned thread has no direct knowledge of it 
    let closure_owned_value: Arc<Mutex<f32>> = Arc::new(Mutex::new(42.0));
    
    // A closure we want to be called by the spawned thread, 
    // but not defined there -- this is considered 'static somehow?
    let on_thread_callback = move |msg: String| {
        println!("Got back message from thread: {:?}, owned value: {:?}", msg, closure_owned_value);
    };

    let (tx, rx) = std::sync::mpsc::channel();

    spawn_thread_with_callback(rx, on_thread_callback);

    tx
}

fn spawn_thread_with_callback<F>(
    rx: Receiver<String>,
    callback: F
) -> std::thread::JoinHandle<()>
where
    F: Fn(String) -> () + std::marker::Send + 'static
{
    std::thread::spawn(move || {
        /* Run an infinite loop as long as channel is open. */
        while let Ok(message) = rx.recv() {
            println!("Thread received message: {:?}", message);
            callback(String::from("Hello back from thread!"));
        }
    })
}

我有少数生产者，他们各自向自己的流发出数据，然后我将它们合并为一个，进一步处理这些值并使用它们：

fun makeInputFlow() = flow {
    while (shouldMakeRequest()) {
        // make (possibly blocking) network request
        // do some CPU-intensive operations
        results.forEach { emit(it) }
        yield()
    }
}

suspend fun runFlows() {
    val flow1 = makeInputFlow()
    val flow2 = makeInputFlow()
    val flow3 = makeInputFlow()

    val merged = listOf(flow1, flow2, flow3)
        .merge()
    merged.collect { println(it) }
}

这是可行的，但看起来只有一个流程实际上在当时产生了值。CPU 大部分时间都处于空闲状态。

有没有办法在自己的协程/线程中安全地运行每个流程，以便所有流程并行产生值？

我正在实现命令模式的实现，其中订阅者将从读取器线程接收数据，并且我在使用 Rust mocks 测试我的实现时遇到了问题。请考虑这个简化的示例：

use std::thread;

pub trait Subscriber: Sync + Send {
    fn callback(&self);
}

pub struct Publisher {
    thread_handler: Option<thread::JoinHandle<()>>,
}

impl Publisher {
    pub fn new() -> Self {
        Self {
            thread_handler: None,
        }
    }

    pub fn start(&mut self, sub: Box<dyn Subscriber>) {
        self.thread_handler = Some(thread::spawn(move || {
            sub.callback();
        }));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use mockall::*;

    mock! {
        TestSubscriber {}

        impl Subscriber for TestSubscriber {
            fn callback(&self);
        }
    }

    #[test]
    fn test_subscriber() {
        let mut publisher = Publisher::new();

        let mut mock_test_subscriber = Box::new(MockTestSubscriber::new());

        publisher.start(mock_test_subscriber);

        mock_test_subscriber.expect_callback().times(1);

        thread::sleep(std::time::Duration::from_millis(10));
    }
}

这会引发

error[E0382]: borrow of moved value: `mock_test_subscriber`
  --> src/foo.rs:46:9
   |
42 |         let mut mock_test_subscriber = Box::new(MockTestSubscriber::new());
   |             ------------------------ move occurs because `mock_test_subscriber` has type `Box<MockTestSubscriber>`, which does not implement the `Copy` trait
43 |
44 |         publisher.start(mock_test_subscriber);
   |                           -------------------- value moved here
45 |
46 |         mock_test_subscriber.expect_callback().times(1);
   |         ^^^^^^^^^^^^^^^^^^^^ value borrowed here after move

我真的不知道如何解决这个问题，我尝试过绕过它dyn Subscriber，Arc但仍然无法让它工作。任何建议都将不胜感激。或者也许有更好的设计来实现相同的结果。