multithreading - 大量多线程操作

Question

使用下面的新代码进行编辑

我在多线程方面相对新手，但为了实现我的目标，快速完成并学习新东西，我决定使用多线程应用程序来完成它。

目标:从文件中解析大量字符串并使用 CoreData 将每个单词保存到 SQLite 数据库中。巨大，因为字数约为 300.000 ...

这就是我的方法。

第1步.将所有单词解析到文件中，并将其放入一个巨大的NSArray中。 (很快就完成了)

第 2 步:创建插入 NSBlockOperation 的 NSOperationQueue。

主要问题是该过程启动得非常快，但很快就会变慢。我正在使用 NSOperationQueue，最大并发操作设置为 100。我有一个 Core 2 Duo 进程(无 HT 的双核)。

我发现使用 NSOperationQueue 创建 NSOperation 会产生大量开销(停止调度队列需要大约 3 分钟才能创建 300k NSOperation。)当我开始调度队列时，CPU 利用率达到 170%。

我也尝试删除 NSOperationQueue 并使用 GDC(300k 循环是瞬时完成的(注释行))，但 cpu 使用率仅为 95%，问题与 NSOperations 相同。很快这个过程就会变慢。

有什么技巧可以做得很好吗？

这里有一些代码(原始问题代码):

- (void)inserdWords:(NSArray *)words insideDictionary:(Dictionary *)dictionary {
    NSDate *creationDate = [NSDate date];

    __block NSUInteger counter = 0;

    NSArray *dictionaryWords = [dictionary.words allObjects];
    NSMutableSet *coreDataWords = [NSMutableSet setWithCapacity:words.count];

    NSLog(@"Begin Adding Operations");

    for (NSString *aWord in words) {

        void(^wordParsingBlock)(void) = ^(void) {
            @synchronized(dictionary) {
                NSManagedObjectContext *context = [(PRDGAppDelegate*)[[NSApplication sharedApplication] delegate] managedObjectContext];                

                [context lock];

                Word *toSaveWord = [NSEntityDescription insertNewObjectForEntityForName:@"Word" inManagedObjectContext:context];
                [toSaveWord setCreated:creationDate];
                [toSaveWord setText:aWord];
                [toSaveWord addDictionariesObject:dictionary];

                [coreDataWords addObject:toSaveWord];
                [dictionary addWordsObject:toSaveWord];

                [context unlock];

                counter++;
                [self.countLabel performSelectorOnMainThread:@selector(setStringValue:) withObject:[NSString stringWithFormat:@"%lu/%lu", counter, words.count] waitUntilDone:NO];

            }
        };

        [_operationsQueue addOperationWithBlock:wordParsingBlock];
//        dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
//        dispatch_async(queue, wordParsingBlock);
    }
    NSLog(@"Operations Added");
}

提前谢谢您。

编辑...

感谢 Stephen Darlington，我重写了代码并解决了问题。最重要的是:不要在线程之间共享 CoreData 对象 ...这意味着不要混合不同上下文检索的 Core 数据对象。

这让我使用 @synchronized(dictionary) 来导致代码执行缓慢!然后我仅使用 MAXTHREAD 实例删除了大量 NSOperation 创建。 (2 或 4 而不是 300k ...是一个巨大的差异)

现在我可以在 30/40 秒内解析 300k+ 字符串。感人的!!我仍然有一些问题(接缝它解析的单词比仅使用 1 个线程的单词要多，并且如果线程超过 1，它不会解析所有单词......我需要弄清楚)但现在代码非常高效。也许下一步可以使用 OpenCL 并将其注入(inject) GPU :)

这是新代码

- (void)insertWords:(NSArray *)words forLanguage:(NSString *)language {
    NSDate *creationDate = [NSDate date];
    NSPersistentStoreCoordinator *coordinator = [(PRDGAppDelegate*)[[NSApplication sharedApplication] delegate] persistentStoreCoordinator];

    // The number of words to be parsed by the single thread.
    NSUInteger wordsPerThread = (NSUInteger)ceil((double)words.count / (double)MAXTHREADS);

    NSLog(@"Start Adding Operations");
    // Here I minimized the number of threads. Every thread will parse and convert a finite number of words instead of 1 word per thread.
    for (NSUInteger threadIdx = 0; threadIdx < MAXTHREADS; threadIdx++) {

        // The NSBlockOperation.
        void(^threadBlock)(void) = ^(void) {
            // A new Context for the current thread.
            NSManagedObjectContext *context = [[NSManagedObjectContext alloc] init];
            [context setPersistentStoreCoordinator:coordinator];

            // Dictionary now is in accordance with the thread context.
            Dictionary *dictionary = [PRDGMainController dictionaryForLanguage:language usingContext:context];

            // Stat Variable. Needed to update the UI.
            NSTimeInterval beginInterval = [[NSDate date] timeIntervalSince1970];
            NSUInteger operationPerInterval = 0;

            // The NSOperation Core. It create a CoreDataWord.
            for (NSUInteger wordIdx = 0; wordIdx < wordsPerThread && wordsPerThread * threadIdx + wordIdx < words.count; wordIdx++) {
                // The String to convert
                NSString *aWord = [words objectAtIndex:wordsPerThread * threadIdx + wordIdx];

                // Some Exceptions to skip certain words.
                if (...) {
                    continue;
                }

                // CoreData Conversion.
                Word *toSaveWord = [NSEntityDescription insertNewObjectForEntityForName:@"Word" inManagedObjectContext:context];
                [toSaveWord setCreated:creationDate];
                [toSaveWord setText:aWord];
                [toSaveWord addDictionariesObject:dictionary];

                operationPerInterval++;

                NSTimeInterval endInterval = [[NSDate date] timeIntervalSince1970];

                // Update case.
                if (endInterval - beginInterval > UPDATE_INTERVAL) {

                    NSLog(@"Thread %lu Processed %lu words", threadIdx, wordIdx);

                    // UI Update. It will be updated only by the first queue.
                    if (threadIdx == 0) {

                        // UI Update code.
                    }
                    beginInterval = endInterval;
                    operationPerInterval = 0;
                }
            }

            // When the NSOperation goes to finish the CoreData thread context is saved.
            [context save:nil];
            NSLog(@"Operation %lu finished", threadIdx);
        };

        // Add the NSBlockOperation to queue.
        [_operationsQueue addOperationWithBlock:threadBlock];
    }
    NSLog(@"Operations Added");
}

Answer 1

一些想法:

• 将最大并发操作数设置得这么高不会产生太大效果。如果您有两个核心，则不可能超过两个
• 看起来您对所有进程都使用相同的 NSManagedObjectContext。这不好
• 假设您的最大并发操作数为 100。瓶颈将是主线程，您将在主线程中尝试更新每个操作的标签。尝试为每n条记录而不是每条记录更新主线程
• 如果您正确使用 Core Data，则不需要锁定上下文...这意味着为每个线程使用不同的上下文
• 您似乎从未保存过上下文？
• 批处理操作是提高性能的好方法...但请参阅上一点
• 正如您所建议的，创建 GCD 操作会产生开销。为每个单词创建一个新单词可能不是最佳选择。您需要平衡创建新进程的开销与并行化的好处

简而言之，即使您使用 GCD 之类的东西，线程也很困难。