Java IRecordProcessorFactory类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory类的典型用法代码示例。如果您正苦于以下问题：Java IRecordProcessorFactory类的具体用法？Java IRecordProcessorFactory怎么用？Java IRecordProcessorFactory使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

IRecordProcessorFactory类属于com.amazonaws.services.kinesis.clientlibrary.interfaces包，在下文中一共展示了IRecordProcessorFactory类的8个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: startKinesisConsumer

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
private void startKinesisConsumer() throws Exception {
    AWSCredentialsProvider credentialsProvider = new
            DefaultAWSCredentialsProviderChain();

    String region = "eu-west-1";
    logger.info("Starting in Region " + region);

    String workerId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + UUID.randomUUID();

    KinesisClientLibConfiguration kinesisClientLibConfiguration = new KinesisClientLibConfiguration(
            this.getClass().getName(), TestConstants.stream, credentialsProvider, workerId)
            .withInitialPositionInStream(InitialPositionInStream.LATEST).withRegionName(region);

    IRecordProcessorFactory recordProcessorFactory = new
            RecordFactory();
    worker = new Worker(recordProcessorFactory,
            kinesisClientLibConfiguration);

    es = Executors.newSingleThreadExecutor();
    es.execute(worker);
}

开发者ID:SaschaMoellering，项目名称:ingestion-service，代码行数:22，代码来源:KafkaKinesisIntegrationTest.java

示例2: KafkaDynamoStreamAdapter

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
public KafkaDynamoStreamAdapter(String regionName, String srcTable, IRecordProcessorFactory processorFactory) {
    sourceTable = srcTable;
    credentialsProvider = new DefaultAWSCredentialsProviderChain();
    recordProcessorFactory = processorFactory;

    adapterClient = new AmazonDynamoDBStreamsAdapterClient(credentialsProvider, new ClientConfiguration());
    dynamoDBClient = new AmazonDynamoDBClient(credentialsProvider, new ClientConfiguration());
    cloudWatchClient = new AmazonCloudWatchClient(credentialsProvider, new ClientConfiguration());

    if ("local".equalsIgnoreCase(regionName)) {
        setClientEndpoints(localddbEndpoint);
    } else if (regionName != null) {
        Region region = Region.getRegion(Regions.fromName(regionName));
        adapterClient.setRegion(region);
        dynamoDBClient.setRegion(region);
        cloudWatchClient.setRegion(region);
    }
}

开发者ID:gnethercutt，项目名称:dynamodb-streams-kafka，代码行数:19，代码来源:KafkaDynamoStreamAdapter.java

示例3: start

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
@Override
public void start() {

    IRecordProcessorFactory recordProcessorFactory = new RecordProcessorFactory(getChannelProcessor(),
                                                                                serializer,
                                                                                backOffTimeInMillis,
                                                                                numberRetries,
                                                                                checkpointIntervalMillis);

    worker = new Worker(recordProcessorFactory, kinesisClientLibConfiguration);

    Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
            System.out.println("Shutting down Kinesis client thread...");
            worker.shutdown();
        }
    });

    try{
        worker.run();
    }catch (AmazonClientException e) {
        logger.error("Can't connect to amazon kinesis", e);
        Throwables.propagate(e);
    }

}

开发者ID:DevOps-TangoMe，项目名称:flume-kinesis，代码行数:27，代码来源:KinesisSource.java

示例4: run

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
public int run() throws Exception {
    configure();

    System.out.println(String.format("Starting %s", appName));
    LOG.info(String.format("Running %s to process stream %s", appName, streamName));

    IRecordProcessorFactory recordProcessorFactory = new ManagedClientProcessorFactory(
            this.templateProcessor);
    Worker worker = new Worker(recordProcessorFactory, this.config);

    int exitCode = 0;
    int failures = 0;

    // run the worker, tolerating as many failures as is configured
    while (failures < failuresToTolerate || failuresToTolerate == -1) {
        try {
            worker.run();
        } catch (Throwable t) {
            LOG.error("Caught throwable while processing data.", t);

            failures++;

            if (failures < failuresToTolerate) {
                LOG.error("Restarting...");
            }
            exitCode = 1;
        }
    }

    return exitCode;
}

开发者ID:awslabs，项目名称:aws-kinesis-beanstalk-workers，代码行数:32，代码来源:ManagedConsumer.java

示例5: main

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
       java.security.Security.setProperty("networkaddress.cache.ttl" , "60");
       
       Server server = new Server();
       ServerConnector connector = new ServerConnector(server);
       connector.setPort(80);
       server.addConnector(connector);

       ResourceHandler resource_handler = new ResourceHandler();
       resource_handler.setResourceBase(".");
       ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
       context.setContextPath("/");
       ServletHolder holderEvents = new ServletHolder("ws-events", MessageProxyServlet.class);
       context.addServlet(holderEvents, "/kinesisapp/*");

       HandlerList handlers = new HandlerList();
       handlers.setHandlers(new Handler[] { resource_handler, context, new DefaultHandler() });
       server.setHandler(handlers);        
       server.start();
       
       AWSCredentialsProvider credentialsProvider = new EnvironmentVariableCredentialsProvider();
       KinesisClientLibConfiguration kinesisConfig = new KinesisClientLibConfiguration(
       		System.getProperty("kinesisapp.name"), 
       		System.getProperty("kinesisapp.stream"), 
       		credentialsProvider, Long.toString(System.currentTimeMillis()))
       		.withKinesisEndpoint(System.getProperty("kinesisapp.endpoint"));
       
       IRecordProcessorFactory factory = new Factory();
       Worker worker = new Worker(factory, kinesisConfig);
       worker.run();
       
       /*
	Processor p = new Processor();
	p.test();
	*/
}

开发者ID:tyagihas，项目名称:awsbigdata，代码行数:37，代码来源:KinesisServer.java

示例6: run

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
public int run() throws Exception {
	configure();

	System.out.println(String.format("Starting %s", appName));
	LOG.info(String.format("Running %s to process stream %s", appName,
			streamName));

	IRecordProcessorFactory recordProcessorFactory = new AggregatorProcessorFactory(
			aggGroup);
	worker = new Worker(recordProcessorFactory, this.config);

	int exitCode = 0;
	int failures = 0;

	// run the worker, tolerating as many failures as is configured
	while (failures < failuresToTolerate || failuresToTolerate == -1) {
		try {
			worker.run();
		} catch (Throwable t) {
			LOG.error("Caught throwable while processing data.", t);

			failures++;

			if (failures < failuresToTolerate) {
				LOG.error("Restarting...");
			} else {
				shutdown();
			}
			exitCode = 1;
		}
	}

	return exitCode;
}

开发者ID:awslabs，项目名称:amazon-kinesis-aggregators，代码行数:35，代码来源:AggregatorConsumer.java

示例7: main

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
/**
 * @param args Property file with config overrides (e.g. application name, stream name)
 * @throws IOException Thrown if we can't read properties from the specified properties file
 */
public static void main(String[] args) throws IOException {
    String propertiesFile = null;

    if (args.length > 1) {
        System.err.println("Usage: java " + KinesisApplication.class.getName() + " <propertiesFile>");
        System.exit(1);
    } else if (args.length == 1) {
        propertiesFile = args[0];
    }

    configure(propertiesFile);

    System.out.println("Starting " + applicationName);
    LOG.info("Running " + applicationName + " to process stream " + streamName);


    IRecordProcessorFactory recordProcessorFactory = new KinesisRecordProcessorFactory(redisEndpoint, redisPort);
    Worker worker = new Worker(recordProcessorFactory, kinesisClientLibConfiguration);
    


    int exitCode = 0;
    try {
        worker.run();
    } catch (Throwable t) {
        LOG.error("Caught throwable while processing data.", t);
        exitCode = 1;
    }
    System.exit(exitCode);
}

开发者ID:awslabs，项目名称:aws-big-data-blog，代码行数:35，代码来源:KinesisApplication.java

示例8: main

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory; //导入依赖的package包/类
/**
 * @param args Property file with config overrides (e.g. application name, stream name)
 * @throws IOException Thrown if we can't read properties from the specified properties file
 */
public static void main(String[] args) throws IOException {
    String propertiesFile = null;
    
    if (args.length > 1) {
        System.err.println("Usage: java " + TweetamoServer.class.getName() + " <propertiesFile>");
        System.exit(1);
    } else if (args.length == 1) {
        propertiesFile = args[0];
    }

    configure(propertiesFile);
    
    System.out.println("Starting " + applicationName);
    LOG.info("Running " + applicationName + " to process stream " + streamName);
    
    
    IRecordProcessorFactory recordProcessorFactory = new TweetamoRecordProcessorFactory();
     Worker worker = new Worker(recordProcessorFactory, kinesisClientLibConfiguration);

    int exitCode = 0;
    try {
        worker.run();
    } catch (Throwable t) {
        LOG.error("Caught throwable while processing data.", t);
        exitCode = 1;
    }
    System.exit(exitCode);
}

开发者ID:dselman，项目名称:tweetamo，代码行数:33，代码来源:TweetamoServer.java

注：本文中的com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。