问题[multithreading](coding)

请考虑以下用 Kotlin 编写的 DCL 示例：

@Volatile
private var property: String? = null

private val lock = ReentrantLock()

fun singletonValue(): String {
    if (property == null) {
        lock.withLock {
            if (property == null) {
                property = "Hello World"
            }
        }
    }
    return property!!
}

它与二十年前讨论的 Java 对应部分非常相似，当时 Java 内存模型随着从 Java 1.4 到 Java 5 的过渡而得到修正。话虽如此，我完全确信上述代码片段在 JVM 上运行时会正常运行。

现在，让我们用lateinit var：

@Volatile
private lateinit var property: String

private val lock = ReentrantLock()

fun singletonValue(): String {
    if (!::property.isInitialized) {
        lock.withLock {
            if (!::property.isInitialized) {
                property = "Hello World"
            }
        }
    }
    return property
}

它在语义上与原始示例等同吗？它是否保持相同的原子性和可见性保证？

背景

我有一个Actor名为akka 的程序Client，它管理http与https服务器的连接。客户端有许多功能，包括ping服务和tokenFetcher服务。

代表client一个服务器与另一个服务器之间的“连接”。其设计目的是允许一个服务器server与另一个服务器聊天。

其过程client如下：

1. 定期检查ping对方server是否在线
2. 如果对方server在线，则执行auth并获取令牌
3. 如果令牌有效，则清除向我们请求的所有呼叫

我正在努力解决的是第 3 步。我想知道如何在线程 (actor) 之间安全地实现这一点。

我尝试过的：

我正在使用Seq客户端存储的消息，如下所示：

case class SendApiCall(apiCall: ApiCall, route: String, var sent: Boolean = false)

class Client(server: Server) extends Actor {
    private var apiCalls: Seq[SendApiCall] = Seq.empty[SendApiCall]

    ...

    override def receive: Receive = {
        case sendApiCall@SendApiCall(_, _, _) =>
            if (server.onlineStatus == OFFLINE) {
                apiCalls = apiCalls.appended(sendApiCall)
            }
            else {
                sendApiCall(sendApiCall)
            }
        
        case ServerOnline() => // <- this is send to us from the ping service when it first detects the server is online
            
            apiCalls.iterator.foreach( apiCallRequest =>
                if (!apiCallRequest.sent) {
                    sendApiCall(apiCallRequest)
                    apiCallRequest.sent = true
                }
                apiCallRequest
            )

            apiCalls = apiCalls.filterNot(apiCallRequest => apiCallRequest.sent)
    }
}

但是，我相信apiCalls这是mutable这种情况的状态吗？我想知道：

1. 这个线程安全吗？
2. 如果不是线程安全的，我该如何让它变得安全？

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

我使用这个链：
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!"));
        }
    })
}