{objccn.io}学习笔记-并发编程-常见的后台实践

在#2 并发编程 中的 常见的后台实践 的学习过程中，

进阶：后台文件 I/O

篇幅中的示例应用中有点小错误，这里纠正一下：

1.读取文件流的四个阶段事件处理代码如下，我做了简单的注解：

#pragma mark- 处理事件流
- (void)stream:(NSStream*)stream handleEvent:(NSStreamEvent)eventCode
{
    switch (eventCode) {
        case NSStreamEventOpenCompleted: {
            NSLog(@"NSStreamEventOpenCompleted");
            // open 打开完成
            break;
        }
        case NSStreamEventEndEncountered: {
            NSLog(@"NSStreamEventEndEncountered");
            // 读取结束
            NSLog(@"waitUntilAllOperationsAreFinished");
            [self.queue waitUntilAllOperationsAreFinished];
            NSLog(@"waitUntilAllOperationsAreFinished >> Finished.");
            [self emitLineWithData:self.remainder];
            self.remainder = nil;
            [self.inputStream close];
            self.inputStream = nil;
            [self.queue addOperationWithBlock:^{
                self.completion(self.lineNumber);
            }];
            break;
        }
        case NSStreamEventErrorOccurred: {
            NSLog(@"NSStreamEventErrorOccurred");
            // 读取错误
            NSLog(@"error"); // TODO
            break;
        }
        case NSStreamEventHasBytesAvailable: {
            NSLog(@"NSStreamEventHasBytesAvailable");
            // 读取数据块过程
            NSMutableData *buffer = [NSMutableData dataWithLength:4 * 1024];
            // 至少读取4 * 1024字节数据，返回实际读取的数据长度
            NSUInteger length = (NSUInteger) [self.inputStream read:[buffer mutableBytes] maxLength:[buffer length]];
            if (0 < length) {
                [buffer setLength:length];
                __weak id weakSelf = self;
                [self.queue addOperationWithBlock:^{
                    // 读取到数据之后交给队列处理
                    [weakSelf processDataChunk:buffer];
                }];
            }
            break;
        }
        default: {
            break;
        }
    }
}

 在NSStreamEventHasBytesAvailable阶段，通过inputStream读取到数据之后，都将数据交由self.queue addOperationWithBlock在子线程中处理（注意：这个线程不会并发执行，是一个串行队列，所以不用担心processDataChunk的读取会错乱），因此有可能在子线程还没有完全处理结束的时候，就已经进入了NSStreamEventEndEncountered阶段，这会造成数据错乱的，因此我们需要做一个简单的处理，那就是当执行到NSStreamEventEndEncountered阶段时，让queue等待所有的block被执行完毕，再执行接下来的扫尾工作。一行代码：

[self.queue waitUntilAllOperationsAreFinished];

 另外这个lineNumber计算的也不太正确，总是比正常的行数大1，也做修正了。

修改过的代码我后续提交到Github上，先粘贴在这里吧：

AppDelegate.m

//
//  AppDelegate.m
//  InputStreamTest
//
//  Created by Chris Eidhof on 06/17/13.
//  Copyright (c) 2013 Chris Eidhof. All rights reserved.
//

#import "AppDelegate.h"
#import "Reader.h"

@interface AppDelegate ()

@property (nonatomic, strong) Reader *reader;
@property (nonatomic, strong) UIButton *button;
@property (nonatomic, strong) UILabel *label;

@end



@implementation AppDelegate

- (BOOL)application:(UIApplication *)application didFinishLaunchingWithOptions:(NSDictionary *)launchOptions
{
    self.window = [[UIWindow alloc] initWithFrame:[[UIScreen mainScreen] bounds]];
    self.window.backgroundColor = [UIColor whiteColor];

    [self addViews];

    [self.window makeKeyAndVisible];
    
    return YES;
}

- (void)addViews
{
    
    UIViewController *controller = [[UIViewController alloc] init];
    self.window.rootViewController = [[UINavigationController alloc] initWithRootViewController:controller];
    
    self.button = [UIButton buttonWithType:UIButtonTypeRoundedRect];
    self.button.frame = CGRectMake(0, 100, 300, 100);
    [self.button addTarget:self action:@selector(import:) forControlEvents:UIControlEventTouchUpInside];
    [self.button setTitle:@"Press Me" forState:UIControlStateNormal];
    [controller.view addSubview:self.button];

    UISlider *slider = [[UISlider alloc] initWithFrame:CGRectMake(0, 220, 300, 64)];
    slider.continuous = YES;
    [slider addTarget:self action:@selector(sliderMoved:) forControlEvents:UIControlEventValueChanged];
    [controller.view addSubview:slider];

    self.label = [[UILabel alloc] initWithFrame:CGRectMake(0, 300, 200, 64)];
    self.label.textAlignment = NSTextAlignmentCenter;
    [controller.view addSubview:self.label];
}

- (void)sliderMoved:(UISlider *)sender;
{
    self.label.text = [NSString stringWithFormat:@"%g", [sender value]];
}

- (void)import:(id)sender
{
    NSURL *fileURL = [[NSBundle mainBundle] URLForResource:@"Clarissa Harlowe" withExtension:@"txt"];
    NSAssert([[NSFileManager defaultManager] fileExistsAtPath:[fileURL path]], @"Please download the sample data");
    
    self.reader = [[Reader alloc] initWithFileAtURL:fileURL];
    [self.reader enumerateLinesWithBlock:^(NSUInteger i, NSString *line){
        if ((i % 2000ull) == 0) {
//            NSLog(@"i: %lu", (unsigned long)i);
//            NSLog(@"i: %lu, line -> %@", (unsigned long)i, line);
            [[NSOperationQueue mainQueue] addOperationWithBlock:^{
                [self.button setTitle:line forState:UIControlStateNormal];
            }];
        }
    } completionHandler:^(NSUInteger numberOfLines){
//        NSLog(@"lines: %lu", (unsigned long)numberOfLines);
        [self.button setTitle:@"Done" forState:UIControlStateNormal];
    }];
    
}

@end

 Reader.m

//
// Created by chris on 6/17/13.
//

#import "Reader.h"
#import "NSData+EnumerateComponents.h"



@interface Reader () <NSStreamDelegate>

@property (nonatomic, strong) NSInputStream* inputStream;
@property (nonatomic, strong) NSURL *fileURL;
@property (nonatomic, copy) NSData *delimiter;
@property (nonatomic, strong) NSMutableData *remainder;
@property (nonatomic, copy) void (^callback) (NSUInteger lineNumber, NSString* line);
@property (nonatomic, copy) void (^completion) (NSUInteger numberOfLines);
@property (nonatomic) NSUInteger lineNumber;
@property (nonatomic, strong) NSOperationQueue *queue;


@end



@implementation Reader

- (void)enumerateLinesWithBlock:(void (^)(NSUInteger lineNumber, NSString *line))block completionHandler:(void (^)(NSUInteger numberOfLines))completion;
{
    // 负责读取数据的自定义队列
    if (self.queue == nil) {
        self.queue = [[NSOperationQueue alloc] init];
        self.queue.maxConcurrentOperationCount = 1;
    }
    NSAssert(self.queue.maxConcurrentOperationCount == 1, @"Queue can't be concurrent.");
    NSAssert(self.inputStream == nil, @"Cannot process multiple input streams in parallel");
    self.callback = block;
    self.completion = completion;
    self.inputStream = [NSInputStream inputStreamWithURL:self.fileURL];
    self.inputStream.delegate = self;
    
    NSAssert([NSRunLoop mainRunLoop] == [NSRunLoop currentRunLoop], @"Must in Main Run Loop.");
    [self.inputStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [self.inputStream open];
}

- (id)initWithFileAtURL:(NSURL *)fileURL;
{
    if (![fileURL isFileURL]) {
        return nil;
    }
    self = [super init];
    if (self) {
        self.fileURL = fileURL;
        self.delimiter = [@"
" dataUsingEncoding:NSUTF8StringEncoding];
    }
    return self;
}

#pragma mark- 处理事件流
- (void)stream:(NSStream*)stream handleEvent:(NSStreamEvent)eventCode
{
    switch (eventCode) {
        case NSStreamEventOpenCompleted: {
            NSLog(@"NSStreamEventOpenCompleted");
            // open 打开完成
            break;
        }
        case NSStreamEventEndEncountered: {
            NSLog(@"NSStreamEventEndEncountered");
            // 读取结束
            NSLog(@"waitUntilAllOperationsAreFinished");
            [self.queue waitUntilAllOperationsAreFinished];
            NSLog(@"waitUntilAllOperationsAreFinished >> Finished.");
            [self emitLineWithData:self.remainder];
            self.remainder = nil;
            [self.inputStream close];
            self.inputStream = nil;
            [self.queue addOperationWithBlock:^{
                self.completion(self.lineNumber);
            }];
            break;
        }
        case NSStreamEventErrorOccurred: {
            NSLog(@"NSStreamEventErrorOccurred");
            // 读取错误
            NSLog(@"error"); // TODO
            break;
        }
        case NSStreamEventHasBytesAvailable: {
            NSLog(@"NSStreamEventHasBytesAvailable");
            // 读取数据块过程
            NSMutableData *buffer = [NSMutableData dataWithLength:4 * 1024];
            // 至少读取4 * 1024字节数据，返回实际读取的数据长度
            NSUInteger length = (NSUInteger) [self.inputStream read:[buffer mutableBytes] maxLength:[buffer length]];
            if (0 < length) {
                [buffer setLength:length];
                __weak id weakSelf = self;
                [self.queue addOperationWithBlock:^{
                    // 读取到数据之后交给队列处理
                    [weakSelf processDataChunk:buffer];
                }];
            }
            break;
        }
        default: {
            break;
        }
    }
}

- (void)processDataChunk:(NSMutableData *)buffer;
{
    if (self.remainder != nil) {
        [self.remainder appendData:buffer];
    } else {
        self.remainder = buffer;
    }
    [self.remainder obj_enumerateComponentsSeparatedBy:self.delimiter usingBlock:^(NSData* component, BOOL last){
        // 读取数据块过程中，遇到一个完整的新行时，做提交处理(!last -> emitLineWithData)
        if (!last) {
            [self emitLineWithData:component];
            // 如果读取到末尾时发现没有换行符，则把剩下的这些暂存，下次读取到块数据时，remainder会appendData进新的块
        } else if (0 < [component length]) {
            self.remainder = [component mutableCopy];
        } else {
            // 有可能末尾刚好是一个换行符
            self.remainder = nil;
        }
        
//        if (last) {
//            NSLog(@"last is true");
//        }
    }];
}

// 读到一行数据
- (void)emitLineWithData:(NSData *)data;
{
    NSUInteger lineNumber = self.lineNumber + 1;
    self.lineNumber = lineNumber;
    if (0 < data.length) {
        // 转换成字符串并回调
        NSString *line = [[NSString alloc] initWithData:data encoding:NSUTF8StringEncoding];
        NSLog(@">> %@", line);
        self.callback(lineNumber, line);
    }
}

@end

NSData+EnumerateComponents.m

//
// Created by chris on 6/17/13.
//

#import "NSData+EnumerateComponents.h"


@implementation NSData (EnumerateComponents)

/**
 *  读取数据
 *
 *  @param delimiter 分隔符
 *  @param block     数据回调
 */
- (void)obj_enumerateComponentsSeparatedBy:(NSData*)delimiter usingBlock:(void (^)(NSData*, BOOL finalBlock) )block
{
    NSUInteger loc = 0;
    while (YES) {
        // 查找新行
        NSRange rangeOfNewline = [self rangeOfData:delimiter options:0 range:NSMakeRange(loc, self.length - loc)];
        if (rangeOfNewline.location == NSNotFound) {
            break;
        }
        
        NSRange rangeWithDelimiter = NSMakeRange(loc, rangeOfNewline.location - loc + delimiter.length);
        NSData *chunkWithDelimiter = [self subdataWithRange:rangeWithDelimiter];
        block(chunkWithDelimiter, NO);
        loc = NSMaxRange(rangeWithDelimiter);
    }
    // 读取剩下的
    NSData *remainder = [self subdataWithRange:NSMakeRange(loc, self.length - loc)];
    block(remainder, YES);
}

@end

 需要调试的话，把那个示例txt（Clarissa Harlowe.txt）中的内容修改一下，简单放置几行即可。