如何將25k記錄放入kinesis流中並使用測試工具來確認它

Question

我開發了一款將記錄寫入Amazon kinesis Stream web服務的軟件。我想了解我們是否有任何軟件工具可以讓我在一秒鐘內測量我的代碼爲Kinesis Stream生成1個碎片的最大吞吐量。 是的，我同意它也取決於硬件配置。但一開始我想Ø知道通用的機器，那麼也許我將能夠看到橫向擴展

有了這個，我想實現25K記錄每秒寫入室壁運動流

參考：室壁運動http://aws.amazon.com/kinesis/

Answer 1

我相信你可以使用Apache JMeter此爲

1. 下載並安裝JMeter的
2. 下載Amazon Kinesis Java Client Library個摔罐子JMeter的classpath中（你可以用你的JMeter安裝的/ lib文件夾）和
3. 使用JSR223 Sampler，「常規」作爲一種語言AmazonKinesisRecordProducerSample作爲參考實現，這將寫代碼記錄到流

請參閱Beanshell vs JSR223 vs Java JMeter Scripting: The Performance-Off You've Been Waiting For!關於安裝「groovy」引擎支持和腳本最佳實踐的指導。

Answer 2

感謝您的提示。我已經想通出路在常規工作代碼使用AWS-Java的SDK使用的Kinesis流發送記錄： 這裏是示例代碼：

/* 
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. 
* 
* Licensed under the Apache License, Version 2.0 (the "License"). 
* You may not use this file except in compliance with the License. 
* A copy of the License is located at 
* 
* http://aws.amazon.com/apache2.0 
* 
* or in the "license" file accompanying this file. This file is distributed 
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either 
* express or implied. See the License for the specific language governing 
* permissions and limitations under the License. 
*/ 

import java.nio.ByteBuffer 
import java.util.List 
import java.util.concurrent.TimeUnit 

import com.amazonaws.AmazonClientException 
import com.amazonaws.AmazonServiceException 
import com.amazonaws.auth.AWSCredentials 
import com.amazonaws.auth.profile.ProfileCredentialsProvider 
import com.amazonaws.services.kinesis.AmazonKinesisClient 
import com.amazonaws.services.kinesis.model.CreateStreamRequest 
import com.amazonaws.services.kinesis.model.DescribeStreamRequest 
import com.amazonaws.services.kinesis.model.DescribeStreamResult 
import com.amazonaws.services.kinesis.model.ListStreamsRequest 
import com.amazonaws.services.kinesis.model.ListStreamsResult 
import com.amazonaws.services.kinesis.model.PutRecordRequest 
import com.amazonaws.services.kinesis.model.PutRecordResult 
import com.amazonaws.services.kinesis.model.ResourceNotFoundException 
import com.amazonaws.services.kinesis.model.StreamDescription 

class AmazonKinesisRecordProducerSample { 

    /* 
    * Before running the code: 
    *  Fill in your AWS access credentials in the provided credentials 
    *  file template, and be sure to move the file to the default location 
    *  (~/.aws/credentials) where the sample code will load the 
    *  credentials from. 
    *  https://console.aws.amazon.com/iam/home?#security_credential 
    * 
    * WARNING: 
    *  To avoid accidental leakage of your credentials, DO NOT keep 
    *  the credentials file in your source directory. 
    */ 

    def kinesis 

    def init() { 
     /* 
     * The ProfileCredentialsProvider will return your [default] 
     * credential profile by reading from the credentials file located at 
     * (~/.aws/credentials). 
     */ 
      AWSCredentials credentials = null 
      credentials = new ProfileCredentialsProvider().getCredentials() 
      kinesis = new AmazonKinesisClient(credentials)   
    }  
} 

def amazonKinesisRecordProducerSample= new AmazonKinesisRecordProducerSample() 
amazonKinesisRecordProducerSample.init() 

    def myStreamName="<KINESIS STREAM NAME>" 

     println("Press CTRL-C to stop.") 
     // Write records to the stream until this program is aborted. 
     while (true) { 
      def createTime = System.currentTimeMillis() 
      def data='<Data IN STRING FORMAT>' 
      def partitionkey="<PARTITION KEY>" 
      def putRecordRequest = new PutRecordRequest() 
      putRecordRequest.setStreamName(myStreamName) 
      putRecordRequest.setData(ByteBuffer.wrap(String.valueOf(data).getBytes())) 
      putRecordRequest.setPartitionKey(partitionkey) 
      def putRecordResult = new PutRecordResult() 
      putRecordResult = amazonKinesisRecordProducerSample.kinesis.putRecord(putRecordRequest) 
      printf("Successfully put record, partition key : %s, ShardID : %s, SequenceNumber : %s.\n", 
        putRecordRequest.getPartitionKey(), 
        putRecordResult.getShardId(), 
        putRecordResult.getSequenceNumber()) 
     } 

注意：如果你有這樣的代碼只會工作Kinesis流已經創建並啓用。如果您需要創建流然後使用它，請參閱aws-java-sdk src文件夾中給出的代碼示例。