SQL Express 2014版调用拉快照复制订阅者
SQL Express 2014 edition call pull Snapshot replication subscribers
我有 Sql Stander 2014 版，它也是分销商和出版商。我在 Sql Stander Edition 上创建了 Subscriber，在其他 Server.Subscriber pull Snapshot Replication 上工作正常，我可以查看数据。
当我使用 2014 Express 版本时，拉取订阅不起作用，因为 2014 Express 版本不支持 SQL Server Agent。
任何人都可以知道在速成版中存在任何其他选项来调用请求订阅模型。否则我们做不到。
这是我用过的代码
public partial class Form1 : Form
{
public Form1()
{
InitializeComponent();
}
private void Form1_Load(object sender, EventArgs e)
{
// createSubscriber();
//disableTheServer2Distributor();
}
public bool CreatePublication( string publisherName, string publicationName, string publicationDbName)
{
ReplicationDatabase publicationDb;
TransPublication publication;
// Create a connection to the Publisher using Windows Authentication.
ServerConnection conn;
conn = new ServerConnection(publisherName);
try
{
// Connect to the Publisher.
conn.Connect();
// Enable the AdventureWorks database for transactional publishing.
publicationDb = new ReplicationDatabase(publicationDbName, conn);
// If the database exists and is not already enabled,
// enable it for transactional publishing.
if (publicationDb.LoadProperties())
{
if (!publicationDb.EnabledTransPublishing)
{
publicationDb.EnabledTransPublishing = true;
}
// If the Log Reader Agent does not exist, create it.
if (!publicationDb.LogReaderAgentExists)
{
// Specify the Windows account under which the agent job runs.
// This account will be used for the local connection to the
// Distributor and all agent connections that use Windows Authentication.
publicationDb.LogReaderAgentProcessSecurity.Login = textBox1.Text ;
publicationDb.LogReaderAgentProcessSecurity.Password = textBox2.Text;
// Explicitly set authentication mode for the Publisher connection
// to the default value of Windows Authentication.
publicationDb.LogReaderAgentPublisherSecurity.WindowsAuthentication = true;
// Create the Log Reader Agent job.
publicationDb.CreateLogReaderAgent();
}
}
else
{
throw new ApplicationException(String.Format(
"The {0} database does not exist at {1}.",
publicationDb, publisherName));
}
// Set the required properties for the transactional publication.
publication = new TransPublication();
publication.ConnectionContext = conn;
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
// Specify a transactional publication (the default).
publication.Type = PublicationType.Snapshot;
// Activate the publication so that we can add subscriptions.
publication.Status = State.Active;
// Enable push and pull subscriptions and independent Distribition Agents.
publication.Attributes |= PublicationAttributes.AllowPull;
//publication.Attributes |= PublicationAttributes.AllowPush;
//publication.Attributes |= PublicationAttributes.IndependentAgent;
// Specify the Windows account under which the Snapshot Agent job runs.
// This account will be used for the local connection to the
// Distributor and all agent connections that use Windows Authentication.
publication.SnapshotGenerationAgentProcessSecurity.Login = textBox1.Text;
publication.SnapshotGenerationAgentProcessSecurity.Password = textBox2.Text;
// Explicitly set the security mode for the Publisher connection
// Windows Authentication (the default).
publication.SnapshotGenerationAgentPublisherSecurity.WindowsAuthentication = true;
ReplicationAgentSchedule schedule;
if (publication.LoadProperties() || publication.SnapshotAvailable)
{
// Set a weekly schedule for the filtered data snapshot.
schedule = new ReplicationAgentSchedule();
schedule.FrequencyType = ScheduleFrequencyType.Continuously;
schedule.FrequencyRecurrenceFactor = 1;
schedule.FrequencyInterval = Convert.ToInt32(0x001);
}
if (!publication.IsExistingObject)
{
// Create the transactional publication.
publication.Create();
// Create a Snapshot Agent job for the publication.
publication.CreateSnapshotAgent();
}
else
{
throw new ApplicationException(String.Format(
"The {0} publication already exists.", publicationName));
}
return true;
}
catch (Exception ex)
{
return false;
// Implement custom application error handling here.
throw new ApplicationException(String.Format(
"The publication {0} could not be created.", publicationName), ex);
}
finally
{
conn.Disconnect();
}
}
/// <summary>
///
/// </summary>
/// <param name="distributionDbName">distribution data base Name</param>
/// <param name="publisherName">Publisher Name </param>
/// <param name="publicationDbName">Publication data base name</param>
/// <param name="distributionDbPassword"> Set the password of the database</param>
/// <param name="WorkingDirectoryPath">Network location from where it can be access </param>
/// <returns></returns>
public bool ConfigurationPublishreAndDistributor(string distributionDbName , string publisherName, string publicationDbName, string distributionDbPassword,string WorkingDirectoryPath)
{
DistributionDatabase distributionDb;
ReplicationServer distributor;
DistributionPublisher publisher;
ReplicationDatabase publicationDb;
// Create a connection to the server using Windows Authentication.
ServerConnection conn = new ServerConnection(publisherName);
try
{
// Connect to the server acting as the Distributor
// and local Publisher.
conn.Connect();
// Define the distribution database at the Distributor,
// but do not create it now.
distributionDb = new DistributionDatabase(distributionDbName, conn);
distributionDb.MaxDistributionRetention = 96;
distributionDb.HistoryRetention = 120;
// Set the Distributor properties and install the Distributor.
// This also creates the specified distribution database.
distributor = new ReplicationServer(conn);
distributor.InstallDistributor(distributionDbPassword, distributionDb);
// Set the Publisher properties and install the Publisher.
publisher = new DistributionPublisher(publisherName, conn);
publisher.DistributionDatabase = distributionDb.Name;
publisher.WorkingDirectory = WorkingDirectoryPath;
publisher.PublisherSecurity.WindowsAuthentication = true;
publisher.Create();
// Enable AdventureWorks2012 as a publication database.
publicationDb = new ReplicationDatabase(publicationDbName, conn);
publicationDb.EnabledTransPublishing = true;
publicationDb.EnabledMergePublishing = true;
return true;
}
catch (Exception ex)
{
// Implement appropriate error handling here.
return false;
throw new ApplicationException("An error occured when installing distribution and publishing.", ex);
}
finally
{
conn.Disconnect();
}
}
private void CreateDistributorServer1_Click(object sender, EventArgs e)
{
if (ConfigurationPublishreAndDistributor("distribution", @"SERVER-002\SQLSERVER2014", "ReplicationDB", "Asdf1234!", @"\SERVER-001\Replication-001"))
{
if (CreatePublication(@"SERVER-002\SQLSERVER2014", "ReplicationSnapShort", "ReplicationDB"))
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationName = "ReplicationSnapShort";
string publicationDbName = "ReplicationDB";
string articleName = "student";
string schemaOwner = "dbo";
TransArticle article;
// Create a connection to the Publisher.
ServerConnection conn = new ServerConnection(publisherName);
// Create a filtered transactional articles in the following steps:
// 1) Create the article with a horizontal filter clause.
// 2) Add columns to or remove columns from the article.
try
{
// Connect to the Publisher.
conn.Connect();
// Define a horizontally filtered, log-based table article.
article = new TransArticle();
article.ConnectionContext = conn;
article.Name = articleName;
article.DatabaseName = publicationDbName;
article.SourceObjectName = articleName;
article.SourceObjectOwner = schemaOwner;
article.PublicationName = publicationName;
article.Type = ArticleOptions.LogBased;
String[] articlecolumns = new String[2];
articlecolumns[0] = "studentid";
articlecolumns[1] = "name";
article.AddReplicatedColumns(articlecolumns);
//article.FilterClause = "DiscontinuedDate IS NULL";
// Ensure that we create the schema owner at the Subscriber.
article.SchemaOption |= CreationScriptOptions.Schema;
if (!article.IsExistingObject)
{
// Create the article.
article.Create();
}
else
{
throw new ApplicationException(String.Format(
"The article {0} already exists in publication {1}.",
articleName, publicationName));
}
// Create an array of column names to remove from the article.
String[] columns = new String[2];
columns[0] = "studentid";
columns[1] = "name";
// Remove the column from the article.
article.AddReplicatedColumns(columns);
// publication.StartSnapshotGenerationAgentJob();
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The article could not be created.", ex);
}
finally
{
conn.Disconnect();
startTheAgentNow();
}
}
}
MessageBox.Show("Create the Distributor and Publisher");
}
public bool startTheAgentNow()
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationName = "ReplicationSnapShort";
string publicationDbName = "ReplicationDB";
TransPublication publication;
// Create a connection to the Publisher using Windows Authentication.
ServerConnection conn;
conn = new ServerConnection(publisherName);
try
{
// Connect to the Publisher.
conn.Connect();
// Set the required properties for an existing publication.
publication = new TransPublication();
publication.ConnectionContext = conn;
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
if (publication.LoadProperties())
{
// Start the Snapshot Agent job for the publication.
publication.StartSnapshotGenerationAgentJob();
}
else
{
throw new ApplicationException(String.Format(
"The {0} publication does not exist.", publicationName));
}
}
catch (Exception ex)
{
// Implement custom application error handling here.
throw new ApplicationException(String.Format(
"A snapshot could not be generated for the {0} publication."
, publicationName), ex);
}
finally
{
conn.Disconnect();
// createSubscriber();
}
return true;
}
public bool createSubscriber()
{
// Define the Publisher, publication, and databases.
string publicationName = "ReplicationSnapShort";
string publisherName = @"SERVER-002\SQLSERVER2014";
string subscriberName = @"SERVER-001\SQLSERVER2014";
string subscriptionDbName = "ReplicationDB1";
string publicationDbName = "ReplicationDB";
//Create connections to the Publisher and Subscriber.
ServerConnection subscriberConn = new ServerConnection(subscriberName);
ServerConnection publisherConn = new ServerConnection(publisherName);
// Create the objects that we need.
TransPublication publication;
TransPullSubscription subscription;
try
{
// Connect to the Publisher and Subscriber.
subscriberConn.Connect();
publisherConn.Connect();
// Ensure that the publication exists and that
// it supports pull subscriptions.
publication = new TransPublication();
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
publication.ConnectionContext = publisherConn;
if (publication.IsExistingObject)
{
if ((publication.Attributes & PublicationAttributes.AllowPull) == 0)
{
publication.Attributes |= PublicationAttributes.AllowPull;
}
// Define the pull subscription.
subscription = new TransPullSubscription();
subscription.ConnectionContext = subscriberConn;
subscription.PublisherName = publisherName;
subscription.PublicationName = publicationName;
subscription.PublicationDBName = publicationDbName;
subscription.DatabaseName = subscriptionDbName;
// Specify the Windows login credentials for the Distribution Agent job.
subscription.SynchronizationAgentProcessSecurity.Login = textBox1.Text;
subscription.SynchronizationAgentProcessSecurity.Password = textBox2.Text;
// Make sure that the agent job for the subscription is created.
subscription.CreateSyncAgentByDefault = true;
// By default, subscriptions to transactional publications are synchronized
// continuously, but in this case we only want to synchronize on demand.
subscription.AgentSchedule.FrequencyType = ScheduleFrequencyType.Continuously;
// Create the pull subscription at the Subscriber.
subscription.Create();
Boolean registered = false;
// Verify that the subscription is not already registered.
foreach (TransSubscription existing
in publication.EnumSubscriptions())
{
if (existing.SubscriberName == subscriberName
&& existing.SubscriptionDBName == subscriptionDbName)
{
registered = true;
}
}
if (!registered)
{
// Register the subscription with the Publisher.
publication.MakePullSubscriptionWellKnown(
subscriberName, subscriptionDbName,
SubscriptionSyncType.Automatic,
TransSubscriberType.ReadOnly);
}
}
else
{
// Do something here if the publication does not exist.
throw new ApplicationException(String.Format(
"The publication '{0}' does not exist on {1}.",
publicationName, publisherName));
}
}
catch (Exception ex)
{
// Implement the appropriate error handling here.
throw new ApplicationException(String.Format(
"The subscription to {0} could not be created.", publicationName), ex);
}
finally
{
subscriberConn.Disconnect();
publisherConn.Disconnect();
}
MessageBox.Show("Subscription is Created");
return true;
}
/// <summary>
/// This is the code of the
/// </summary>
public void disableTheServer2Distributor()
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationDbName = "ReplicationDB";
string distributorName = @"SERVER-002\SQLSERVER2014";
string distributionDbName = "distribution";
// Create connections to the Publisher and Distributor
// using Windows Authentication.
ServerConnection publisherConn = new ServerConnection(publisherName);
ServerConnection distributorConn = new ServerConnection(distributorName);
// Create the objects we need.
ReplicationServer distributor =
new ReplicationServer(distributorConn);
DistributionPublisher publisher;
DistributionDatabase distributionDb =
new DistributionDatabase(distributionDbName, distributorConn);
ReplicationDatabase publicationDb;
publicationDb = new ReplicationDatabase(publicationDbName, publisherConn);
try
{
// Connect to the Publisher and Distributor.
publisherConn.Connect();
distributorConn.Connect();
// Disable all publishing on the AdventureWorks2012 database.
if (publicationDb.LoadProperties())
{
if (publicationDb.EnabledMergePublishing)
{
publicationDb.EnabledMergePublishing = false;
}
else if (publicationDb.EnabledTransPublishing)
{
publicationDb.EnabledTransPublishing = false;
}
}
else
{
throw new ApplicationException(
String.Format("The {0} database does not exist.", publicationDbName));
}
// We cannot uninstall the Publisher if there are still Subscribers.
if (distributor.RegisteredSubscribers.Count == 0)
{
// Uninstall the Publisher, if it exists.
publisher = new DistributionPublisher(publisherName, distributorConn);
if (publisher.LoadProperties())
{
publisher.Remove(false);
}
else
{
// Do something here if the Publisher does not exist.
throw new ApplicationException(String.Format(
"{0} is not a Publisher for {1}.", publisherName, distributorName));
}
// Drop the distribution database.
if (distributionDb.LoadProperties())
{
distributionDb.Remove();
}
else
{
// Do something here if the distribition DB does not exist.
throw new ApplicationException(String.Format(
"The distribution database '{0}' does not exist on {1}.",
distributionDbName, distributorName));
}
// Uninstall the Distributor, if it exists.
if (distributor.LoadProperties())
{
// Passing a value of false means that the Publisher
// and distribution databases must already be uninstalled,
// and that no local databases be enabled for publishing.
distributor.UninstallDistributor(false);
}
else
{
//Do something here if the distributor does not exist.
throw new ApplicationException(String.Format(
"The Distributor '{0}' does not exist.", distributorName));
}
}
else
{
throw new ApplicationException("You must first delete all subscriptions.");
}
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The Publisher and Distributor could not be uninstalled", ex);
}
finally
{
publisherConn.Disconnect();
distributorConn.Disconnect();
}
}
private void CreateDistributorServer2_Click(object sender, EventArgs e)
{
}
private void button5_Click(object sender, EventArgs e)
{
createSubscriber();
}
private void button13_Click(object sender, EventArgs e)
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationDbName = "ReplicationDB";
String subscriberName = @"SERVER-001\SQLSERVER2014";
String publicationName = "ReplicationSnapShort";
String subscriptionDbName = "ReplicationDB1";
// Create a connection to the Subscriber.
ServerConnection conn = new ServerConnection(subscriberName);
TransPullSubscription subscription;
try
{
// Connect to the Subscriber.
conn.Connect();
// Define subscription properties.
subscription = new TransPullSubscription();
subscription.ConnectionContext = conn;
subscription.DatabaseName = subscriptionDbName;
subscription.PublisherName = publisherName;
subscription.PublicationDBName = publicationDbName;
subscription.PublicationName = publicationName;
// If the pull subscription and the job exists, mark the subscription
// for reinitialization and start the agent job.
if (subscription.LoadProperties() && subscription.AgentJobId != null)
{
subscription.Reinitialize();
subscription.SynchronizeWithJob();
}
else
{
// Do something here if the subscription does not exist.
throw new ApplicationException(String.Format(
"A subscription to '{0}' does not exists on {1}",
publicationName, subscriberName));
}
}
catch (Exception ex)
{
// Do appropriate error handling here.
//throw new ApplicationException("The subscription could not be reinitialized.", ex);
}
finally
{
conn.Disconnect();
startTheAgentNow();
MessageBox.Show("Agents are started");
}
}
private void button9_Click(object sender, EventArgs e)
{
}
}
以上代码按预期工作 当我将订户更改为 SQL
时出现问题
我看到您正在使用 TransPullSubscription class 并调用 SynchronizeWithJob() 方法来同步订阅。正如您所发现的，SQL Server Agent 在 SQL Server Express 中不可用，因此此方法不起作用。
但是，RMO TransSynchronizationAgent class exposes a Synchronize method which can be used to synchronize a pull subscription without an agent job. This is covered in How to: Synchronize a Pull Subscription (RMO Programming)。
Get an instance of the TransSynchronizationAgent class from the
SynchronizationAgent property, and call the Synchronize method. This
method starts the agent synchronously, and control remains with the
running agent job. During synchronous execution, you can handle the
Status event while the agent is running.
我有一个类似的合并请求订阅示例 here。
请注意，如果您在创建请求订阅时为 CreateSyncAgentByDefault（默认值）指定了 false 值，则在调用 Synchronize() 之前还需要设置其他属性。
If you specified a value of false for CreateSyncAgentByDefault (the
default) when you created the pull subscription, you also need to
specify Distributor, DistributorSecurityMode, and optionally
DistributorLogin and DistributorPassword because the agent job related
metadata for the subscription is not available in
MSsubscription_properties.
谢谢
布兰登·威廉姆斯
我对代码进行的更改如下
`
{
TransSynchronizationAgent agent;
// Sync BackgroundWorker
BackgroundWorker syncBackgroundWorker;
public Form1()
{
InitializeComponent();
lblSubscriptionName.Text = "[" + subscriptionDbName + "] - [" + publisherName + "] - [" + publicationDbName + "]";
lblPublicationName.Text = publicationName;
}
private void btnStart_Click(object sender, EventArgs e)
{
// Instantiate a BackgroundWorker, subscribe to its events, and call RunWorkerAsync()
syncBackgroundWorker = new BackgroundWorker();
syncBackgroundWorker.WorkerReportsProgress = true;
syncBackgroundWorker.DoWork += new DoWorkEventHandler(syncBackgroundWorker_DoWork);
syncBackgroundWorker.ProgressChanged += new ProgressChangedEventHandler(syncBackgroundWorker_ProgressChanged);
syncBackgroundWorker.RunWorkerCompleted += new RunWorkerCompletedEventHandler(syncBackgroundWorker_RunWorkerCompleted);
// Disable the start button
btnStart.Enabled = false;
// Initialize the progress bar and status textbox
pbStatus.Value = 0;
tbLastStatusMessage.Text = String.Empty;
pictureBoxStatus.Visible = true;
pictureBoxStatus.Enabled = true;
// Kick off a background operation to synchronize
syncBackgroundWorker.RunWorkerAsync();
}
// This event handler initiates the synchronization
private void syncBackgroundWorker_DoWork(object sender, DoWorkEventArgs e)
{
// Connect to the Subscriber and synchronize
SynchronizeMergePullSubscriptionViaRMO();
}
// Synchronize
public void SynchronizeMergePullSubscriptionViaRMO()
{
// Create a connection to the Subscriber.
ServerConnection conn = new ServerConnection(subscriberName);
// Merge pull subscription
TransPullSubscription subscription;
try
{
// Connect to the Subscriber.
conn.Connect();
// Define the pull subscription.
subscription = new TransPullSubscription();
subscription.ConnectionContext = conn;
subscription.DatabaseName = subscriptionDbName;
subscription.PublisherName = publisherName;
subscription.PublicationDBName = publicationDbName;
subscription.PublicationName = publicationName;
// If the pull subscription exists, then start the synchronization.
if (subscription.LoadProperties())
{
// Get the agent for the subscription.
agent = subscription.SynchronizationAgent;
// Set the required properties that could not be returned
// from the MSsubscription_properties table.
//agent.PublisherSecurityMode = SecurityMode.Integrated;
agent.DistributorSecurityMode = SecurityMode.Integrated;
agent.Distributor = publisherName;
// Enable verbose merge agent output to file.
agent.OutputVerboseLevel = 4;
agent.Output = "C:\TEMP\mergeagent.log";
// Handle the Status event
agent.Status += new AgentCore.StatusEventHandler(agent_Status);
// Synchronously start the Merge Agent for the subscription.
agent.Synchronize();
}
else
{
// Do something here if the pull subscription does not exist.
throw new ApplicationException(String.Format(
"A subscription to '{0}' does not exist on {1}",
publicationName, subscriberName));
}
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The subscription could not be " +
"synchronized. Verify that the subscription has " +
"been defined correctly.", ex);
}
finally
{
conn.Disconnect();
}
}
// This event handler handles the Status event and reports the agent progress.
public void agent_Status(object sender, StatusEventArgs e)
{
syncBackgroundWorker.ReportProgress(Convert.ToInt32(e.PercentCompleted), e.Message.ToString());
}
// This event handler updates the form with agent progress.
private void syncBackgroundWorker_ProgressChanged(object sender, ProgressChangedEventArgs e)
{
// Set the progress bar percent completed
pbStatus.Value = e.ProgressPercentage;
// Append the last agent message
tbLastStatusMessage.Text += e.UserState.ToString() + Environment.NewLine;
// Scroll to end
ScrollToEnd();
}
// This event handler deals with the results of the background operation.
private void syncBackgroundWorker_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
{
if (e.Cancelled == true)
{
tbLastStatusMessage.Text += "Canceled!" + Environment.NewLine;
ScrollToEnd();
}
else if (e.Error != null)
{
tbLastStatusMessage.Text += "Error: " + e.Error.Message + Environment.NewLine;
ScrollToEnd();
}
else
{
tbLastStatusMessage.Text += "Done!" + Environment.NewLine;
ScrollToEnd();
}
btnStart.Enabled = true;
pictureBoxStatus.Enabled = false;
}
private void ScrollToEnd()
{
// Scroll to end
tbLastStatusMessage.SelectionStart = tbLastStatusMessage.TextLength;
tbLastStatusMessage.ScrollToCaret();
}
private void btnClose_Click(object sender, EventArgs e)
{
this.Close();
}
}`
我有 Sql Stander 2014 版，它也是分销商和出版商。我在 Sql Stander Edition 上创建了 Subscriber，在其他 Server.Subscriber pull Snapshot Replication 上工作正常，我可以查看数据。
当我使用 2014 Express 版本时，拉取订阅不起作用，因为 2014 Express 版本不支持 SQL Server Agent。
任何人都可以知道在速成版中存在任何其他选项来调用请求订阅模型。否则我们做不到。
这是我用过的代码
public partial class Form1 : Form
{
public Form1()
{
InitializeComponent();
}
private void Form1_Load(object sender, EventArgs e)
{
// createSubscriber();
//disableTheServer2Distributor();
}
public bool CreatePublication( string publisherName, string publicationName, string publicationDbName)
{
ReplicationDatabase publicationDb;
TransPublication publication;
// Create a connection to the Publisher using Windows Authentication.
ServerConnection conn;
conn = new ServerConnection(publisherName);
try
{
// Connect to the Publisher.
conn.Connect();
// Enable the AdventureWorks database for transactional publishing.
publicationDb = new ReplicationDatabase(publicationDbName, conn);
// If the database exists and is not already enabled,
// enable it for transactional publishing.
if (publicationDb.LoadProperties())
{
if (!publicationDb.EnabledTransPublishing)
{
publicationDb.EnabledTransPublishing = true;
}
// If the Log Reader Agent does not exist, create it.
if (!publicationDb.LogReaderAgentExists)
{
// Specify the Windows account under which the agent job runs.
// This account will be used for the local connection to the
// Distributor and all agent connections that use Windows Authentication.
publicationDb.LogReaderAgentProcessSecurity.Login = textBox1.Text ;
publicationDb.LogReaderAgentProcessSecurity.Password = textBox2.Text;
// Explicitly set authentication mode for the Publisher connection
// to the default value of Windows Authentication.
publicationDb.LogReaderAgentPublisherSecurity.WindowsAuthentication = true;
// Create the Log Reader Agent job.
publicationDb.CreateLogReaderAgent();
}
}
else
{
throw new ApplicationException(String.Format(
"The {0} database does not exist at {1}.",
publicationDb, publisherName));
}
// Set the required properties for the transactional publication.
publication = new TransPublication();
publication.ConnectionContext = conn;
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
// Specify a transactional publication (the default).
publication.Type = PublicationType.Snapshot;
// Activate the publication so that we can add subscriptions.
publication.Status = State.Active;
// Enable push and pull subscriptions and independent Distribition Agents.
publication.Attributes |= PublicationAttributes.AllowPull;
//publication.Attributes |= PublicationAttributes.AllowPush;
//publication.Attributes |= PublicationAttributes.IndependentAgent;
// Specify the Windows account under which the Snapshot Agent job runs.
// This account will be used for the local connection to the
// Distributor and all agent connections that use Windows Authentication.
publication.SnapshotGenerationAgentProcessSecurity.Login = textBox1.Text;
publication.SnapshotGenerationAgentProcessSecurity.Password = textBox2.Text;
// Explicitly set the security mode for the Publisher connection
// Windows Authentication (the default).
publication.SnapshotGenerationAgentPublisherSecurity.WindowsAuthentication = true;
ReplicationAgentSchedule schedule;
if (publication.LoadProperties() || publication.SnapshotAvailable)
{
// Set a weekly schedule for the filtered data snapshot.
schedule = new ReplicationAgentSchedule();
schedule.FrequencyType = ScheduleFrequencyType.Continuously;
schedule.FrequencyRecurrenceFactor = 1;
schedule.FrequencyInterval = Convert.ToInt32(0x001);
}
if (!publication.IsExistingObject)
{
// Create the transactional publication.
publication.Create();
// Create a Snapshot Agent job for the publication.
publication.CreateSnapshotAgent();
}
else
{
throw new ApplicationException(String.Format(
"The {0} publication already exists.", publicationName));
}
return true;
}
catch (Exception ex)
{
return false;
// Implement custom application error handling here.
throw new ApplicationException(String.Format(
"The publication {0} could not be created.", publicationName), ex);
}
finally
{
conn.Disconnect();
}
}
/// <summary>
///
/// </summary>
/// <param name="distributionDbName">distribution data base Name</param>
/// <param name="publisherName">Publisher Name </param>
/// <param name="publicationDbName">Publication data base name</param>
/// <param name="distributionDbPassword"> Set the password of the database</param>
/// <param name="WorkingDirectoryPath">Network location from where it can be access </param>
/// <returns></returns>
public bool ConfigurationPublishreAndDistributor(string distributionDbName , string publisherName, string publicationDbName, string distributionDbPassword,string WorkingDirectoryPath)
{
DistributionDatabase distributionDb;
ReplicationServer distributor;
DistributionPublisher publisher;
ReplicationDatabase publicationDb;
// Create a connection to the server using Windows Authentication.
ServerConnection conn = new ServerConnection(publisherName);
try
{
// Connect to the server acting as the Distributor
// and local Publisher.
conn.Connect();
// Define the distribution database at the Distributor,
// but do not create it now.
distributionDb = new DistributionDatabase(distributionDbName, conn);
distributionDb.MaxDistributionRetention = 96;
distributionDb.HistoryRetention = 120;
// Set the Distributor properties and install the Distributor.
// This also creates the specified distribution database.
distributor = new ReplicationServer(conn);
distributor.InstallDistributor(distributionDbPassword, distributionDb);
// Set the Publisher properties and install the Publisher.
publisher = new DistributionPublisher(publisherName, conn);
publisher.DistributionDatabase = distributionDb.Name;
publisher.WorkingDirectory = WorkingDirectoryPath;
publisher.PublisherSecurity.WindowsAuthentication = true;
publisher.Create();
// Enable AdventureWorks2012 as a publication database.
publicationDb = new ReplicationDatabase(publicationDbName, conn);
publicationDb.EnabledTransPublishing = true;
publicationDb.EnabledMergePublishing = true;
return true;
}
catch (Exception ex)
{
// Implement appropriate error handling here.
return false;
throw new ApplicationException("An error occured when installing distribution and publishing.", ex);
}
finally
{
conn.Disconnect();
}
}
private void CreateDistributorServer1_Click(object sender, EventArgs e)
{
if (ConfigurationPublishreAndDistributor("distribution", @"SERVER-002\SQLSERVER2014", "ReplicationDB", "Asdf1234!", @"\SERVER-001\Replication-001"))
{
if (CreatePublication(@"SERVER-002\SQLSERVER2014", "ReplicationSnapShort", "ReplicationDB"))
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationName = "ReplicationSnapShort";
string publicationDbName = "ReplicationDB";
string articleName = "student";
string schemaOwner = "dbo";
TransArticle article;
// Create a connection to the Publisher.
ServerConnection conn = new ServerConnection(publisherName);
// Create a filtered transactional articles in the following steps:
// 1) Create the article with a horizontal filter clause.
// 2) Add columns to or remove columns from the article.
try
{
// Connect to the Publisher.
conn.Connect();
// Define a horizontally filtered, log-based table article.
article = new TransArticle();
article.ConnectionContext = conn;
article.Name = articleName;
article.DatabaseName = publicationDbName;
article.SourceObjectName = articleName;
article.SourceObjectOwner = schemaOwner;
article.PublicationName = publicationName;
article.Type = ArticleOptions.LogBased;
String[] articlecolumns = new String[2];
articlecolumns[0] = "studentid";
articlecolumns[1] = "name";
article.AddReplicatedColumns(articlecolumns);
//article.FilterClause = "DiscontinuedDate IS NULL";
// Ensure that we create the schema owner at the Subscriber.
article.SchemaOption |= CreationScriptOptions.Schema;
if (!article.IsExistingObject)
{
// Create the article.
article.Create();
}
else
{
throw new ApplicationException(String.Format(
"The article {0} already exists in publication {1}.",
articleName, publicationName));
}
// Create an array of column names to remove from the article.
String[] columns = new String[2];
columns[0] = "studentid";
columns[1] = "name";
// Remove the column from the article.
article.AddReplicatedColumns(columns);
// publication.StartSnapshotGenerationAgentJob();
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The article could not be created.", ex);
}
finally
{
conn.Disconnect();
startTheAgentNow();
}
}
}
MessageBox.Show("Create the Distributor and Publisher");
}
public bool startTheAgentNow()
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationName = "ReplicationSnapShort";
string publicationDbName = "ReplicationDB";
TransPublication publication;
// Create a connection to the Publisher using Windows Authentication.
ServerConnection conn;
conn = new ServerConnection(publisherName);
try
{
// Connect to the Publisher.
conn.Connect();
// Set the required properties for an existing publication.
publication = new TransPublication();
publication.ConnectionContext = conn;
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
if (publication.LoadProperties())
{
// Start the Snapshot Agent job for the publication.
publication.StartSnapshotGenerationAgentJob();
}
else
{
throw new ApplicationException(String.Format(
"The {0} publication does not exist.", publicationName));
}
}
catch (Exception ex)
{
// Implement custom application error handling here.
throw new ApplicationException(String.Format(
"A snapshot could not be generated for the {0} publication."
, publicationName), ex);
}
finally
{
conn.Disconnect();
// createSubscriber();
}
return true;
}
public bool createSubscriber()
{
// Define the Publisher, publication, and databases.
string publicationName = "ReplicationSnapShort";
string publisherName = @"SERVER-002\SQLSERVER2014";
string subscriberName = @"SERVER-001\SQLSERVER2014";
string subscriptionDbName = "ReplicationDB1";
string publicationDbName = "ReplicationDB";
//Create connections to the Publisher and Subscriber.
ServerConnection subscriberConn = new ServerConnection(subscriberName);
ServerConnection publisherConn = new ServerConnection(publisherName);
// Create the objects that we need.
TransPublication publication;
TransPullSubscription subscription;
try
{
// Connect to the Publisher and Subscriber.
subscriberConn.Connect();
publisherConn.Connect();
// Ensure that the publication exists and that
// it supports pull subscriptions.
publication = new TransPublication();
publication.Name = publicationName;
publication.DatabaseName = publicationDbName;
publication.ConnectionContext = publisherConn;
if (publication.IsExistingObject)
{
if ((publication.Attributes & PublicationAttributes.AllowPull) == 0)
{
publication.Attributes |= PublicationAttributes.AllowPull;
}
// Define the pull subscription.
subscription = new TransPullSubscription();
subscription.ConnectionContext = subscriberConn;
subscription.PublisherName = publisherName;
subscription.PublicationName = publicationName;
subscription.PublicationDBName = publicationDbName;
subscription.DatabaseName = subscriptionDbName;
// Specify the Windows login credentials for the Distribution Agent job.
subscription.SynchronizationAgentProcessSecurity.Login = textBox1.Text;
subscription.SynchronizationAgentProcessSecurity.Password = textBox2.Text;
// Make sure that the agent job for the subscription is created.
subscription.CreateSyncAgentByDefault = true;
// By default, subscriptions to transactional publications are synchronized
// continuously, but in this case we only want to synchronize on demand.
subscription.AgentSchedule.FrequencyType = ScheduleFrequencyType.Continuously;
// Create the pull subscription at the Subscriber.
subscription.Create();
Boolean registered = false;
// Verify that the subscription is not already registered.
foreach (TransSubscription existing
in publication.EnumSubscriptions())
{
if (existing.SubscriberName == subscriberName
&& existing.SubscriptionDBName == subscriptionDbName)
{
registered = true;
}
}
if (!registered)
{
// Register the subscription with the Publisher.
publication.MakePullSubscriptionWellKnown(
subscriberName, subscriptionDbName,
SubscriptionSyncType.Automatic,
TransSubscriberType.ReadOnly);
}
}
else
{
// Do something here if the publication does not exist.
throw new ApplicationException(String.Format(
"The publication '{0}' does not exist on {1}.",
publicationName, publisherName));
}
}
catch (Exception ex)
{
// Implement the appropriate error handling here.
throw new ApplicationException(String.Format(
"The subscription to {0} could not be created.", publicationName), ex);
}
finally
{
subscriberConn.Disconnect();
publisherConn.Disconnect();
}
MessageBox.Show("Subscription is Created");
return true;
}
/// <summary>
/// This is the code of the
/// </summary>
public void disableTheServer2Distributor()
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationDbName = "ReplicationDB";
string distributorName = @"SERVER-002\SQLSERVER2014";
string distributionDbName = "distribution";
// Create connections to the Publisher and Distributor
// using Windows Authentication.
ServerConnection publisherConn = new ServerConnection(publisherName);
ServerConnection distributorConn = new ServerConnection(distributorName);
// Create the objects we need.
ReplicationServer distributor =
new ReplicationServer(distributorConn);
DistributionPublisher publisher;
DistributionDatabase distributionDb =
new DistributionDatabase(distributionDbName, distributorConn);
ReplicationDatabase publicationDb;
publicationDb = new ReplicationDatabase(publicationDbName, publisherConn);
try
{
// Connect to the Publisher and Distributor.
publisherConn.Connect();
distributorConn.Connect();
// Disable all publishing on the AdventureWorks2012 database.
if (publicationDb.LoadProperties())
{
if (publicationDb.EnabledMergePublishing)
{
publicationDb.EnabledMergePublishing = false;
}
else if (publicationDb.EnabledTransPublishing)
{
publicationDb.EnabledTransPublishing = false;
}
}
else
{
throw new ApplicationException(
String.Format("The {0} database does not exist.", publicationDbName));
}
// We cannot uninstall the Publisher if there are still Subscribers.
if (distributor.RegisteredSubscribers.Count == 0)
{
// Uninstall the Publisher, if it exists.
publisher = new DistributionPublisher(publisherName, distributorConn);
if (publisher.LoadProperties())
{
publisher.Remove(false);
}
else
{
// Do something here if the Publisher does not exist.
throw new ApplicationException(String.Format(
"{0} is not a Publisher for {1}.", publisherName, distributorName));
}
// Drop the distribution database.
if (distributionDb.LoadProperties())
{
distributionDb.Remove();
}
else
{
// Do something here if the distribition DB does not exist.
throw new ApplicationException(String.Format(
"The distribution database '{0}' does not exist on {1}.",
distributionDbName, distributorName));
}
// Uninstall the Distributor, if it exists.
if (distributor.LoadProperties())
{
// Passing a value of false means that the Publisher
// and distribution databases must already be uninstalled,
// and that no local databases be enabled for publishing.
distributor.UninstallDistributor(false);
}
else
{
//Do something here if the distributor does not exist.
throw new ApplicationException(String.Format(
"The Distributor '{0}' does not exist.", distributorName));
}
}
else
{
throw new ApplicationException("You must first delete all subscriptions.");
}
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The Publisher and Distributor could not be uninstalled", ex);
}
finally
{
publisherConn.Disconnect();
distributorConn.Disconnect();
}
}
private void CreateDistributorServer2_Click(object sender, EventArgs e)
{
}
private void button5_Click(object sender, EventArgs e)
{
createSubscriber();
}
private void button13_Click(object sender, EventArgs e)
{
string publisherName = @"SERVER-002\SQLSERVER2014";
string publicationDbName = "ReplicationDB";
String subscriberName = @"SERVER-001\SQLSERVER2014";
String publicationName = "ReplicationSnapShort";
String subscriptionDbName = "ReplicationDB1";
// Create a connection to the Subscriber.
ServerConnection conn = new ServerConnection(subscriberName);
TransPullSubscription subscription;
try
{
// Connect to the Subscriber.
conn.Connect();
// Define subscription properties.
subscription = new TransPullSubscription();
subscription.ConnectionContext = conn;
subscription.DatabaseName = subscriptionDbName;
subscription.PublisherName = publisherName;
subscription.PublicationDBName = publicationDbName;
subscription.PublicationName = publicationName;
// If the pull subscription and the job exists, mark the subscription
// for reinitialization and start the agent job.
if (subscription.LoadProperties() && subscription.AgentJobId != null)
{
subscription.Reinitialize();
subscription.SynchronizeWithJob();
}
else
{
// Do something here if the subscription does not exist.
throw new ApplicationException(String.Format(
"A subscription to '{0}' does not exists on {1}",
publicationName, subscriberName));
}
}
catch (Exception ex)
{
// Do appropriate error handling here.
//throw new ApplicationException("The subscription could not be reinitialized.", ex);
}
finally
{
conn.Disconnect();
startTheAgentNow();
MessageBox.Show("Agents are started");
}
}
private void button9_Click(object sender, EventArgs e)
{
}
}
以上代码按预期工作 当我将订户更改为 SQL
时出现问题我看到您正在使用 TransPullSubscription class 并调用 SynchronizeWithJob() 方法来同步订阅。正如您所发现的，SQL Server Agent 在 SQL Server Express 中不可用，因此此方法不起作用。
但是，RMO TransSynchronizationAgent class exposes a Synchronize method which can be used to synchronize a pull subscription without an agent job. This is covered in How to: Synchronize a Pull Subscription (RMO Programming)。
Get an instance of the TransSynchronizationAgent class from the SynchronizationAgent property, and call the Synchronize method. This method starts the agent synchronously, and control remains with the running agent job. During synchronous execution, you can handle the Status event while the agent is running.
我有一个类似的合并请求订阅示例 here。
请注意，如果您在创建请求订阅时为 CreateSyncAgentByDefault（默认值）指定了 false 值，则在调用 Synchronize() 之前还需要设置其他属性。
If you specified a value of false for CreateSyncAgentByDefault (the default) when you created the pull subscription, you also need to specify Distributor, DistributorSecurityMode, and optionally DistributorLogin and DistributorPassword because the agent job related metadata for the subscription is not available in MSsubscription_properties.
谢谢 布兰登·威廉姆斯 我对代码进行的更改如下
` {
TransSynchronizationAgent agent;
// Sync BackgroundWorker
BackgroundWorker syncBackgroundWorker;
public Form1()
{
InitializeComponent();
lblSubscriptionName.Text = "[" + subscriptionDbName + "] - [" + publisherName + "] - [" + publicationDbName + "]";
lblPublicationName.Text = publicationName;
}
private void btnStart_Click(object sender, EventArgs e)
{
// Instantiate a BackgroundWorker, subscribe to its events, and call RunWorkerAsync()
syncBackgroundWorker = new BackgroundWorker();
syncBackgroundWorker.WorkerReportsProgress = true;
syncBackgroundWorker.DoWork += new DoWorkEventHandler(syncBackgroundWorker_DoWork);
syncBackgroundWorker.ProgressChanged += new ProgressChangedEventHandler(syncBackgroundWorker_ProgressChanged);
syncBackgroundWorker.RunWorkerCompleted += new RunWorkerCompletedEventHandler(syncBackgroundWorker_RunWorkerCompleted);
// Disable the start button
btnStart.Enabled = false;
// Initialize the progress bar and status textbox
pbStatus.Value = 0;
tbLastStatusMessage.Text = String.Empty;
pictureBoxStatus.Visible = true;
pictureBoxStatus.Enabled = true;
// Kick off a background operation to synchronize
syncBackgroundWorker.RunWorkerAsync();
}
// This event handler initiates the synchronization
private void syncBackgroundWorker_DoWork(object sender, DoWorkEventArgs e)
{
// Connect to the Subscriber and synchronize
SynchronizeMergePullSubscriptionViaRMO();
}
// Synchronize
public void SynchronizeMergePullSubscriptionViaRMO()
{
// Create a connection to the Subscriber.
ServerConnection conn = new ServerConnection(subscriberName);
// Merge pull subscription
TransPullSubscription subscription;
try
{
// Connect to the Subscriber.
conn.Connect();
// Define the pull subscription.
subscription = new TransPullSubscription();
subscription.ConnectionContext = conn;
subscription.DatabaseName = subscriptionDbName;
subscription.PublisherName = publisherName;
subscription.PublicationDBName = publicationDbName;
subscription.PublicationName = publicationName;
// If the pull subscription exists, then start the synchronization.
if (subscription.LoadProperties())
{
// Get the agent for the subscription.
agent = subscription.SynchronizationAgent;
// Set the required properties that could not be returned
// from the MSsubscription_properties table.
//agent.PublisherSecurityMode = SecurityMode.Integrated;
agent.DistributorSecurityMode = SecurityMode.Integrated;
agent.Distributor = publisherName;
// Enable verbose merge agent output to file.
agent.OutputVerboseLevel = 4;
agent.Output = "C:\TEMP\mergeagent.log";
// Handle the Status event
agent.Status += new AgentCore.StatusEventHandler(agent_Status);
// Synchronously start the Merge Agent for the subscription.
agent.Synchronize();
}
else
{
// Do something here if the pull subscription does not exist.
throw new ApplicationException(String.Format(
"A subscription to '{0}' does not exist on {1}",
publicationName, subscriberName));
}
}
catch (Exception ex)
{
// Implement appropriate error handling here.
throw new ApplicationException("The subscription could not be " +
"synchronized. Verify that the subscription has " +
"been defined correctly.", ex);
}
finally
{
conn.Disconnect();
}
}
// This event handler handles the Status event and reports the agent progress.
public void agent_Status(object sender, StatusEventArgs e)
{
syncBackgroundWorker.ReportProgress(Convert.ToInt32(e.PercentCompleted), e.Message.ToString());
}
// This event handler updates the form with agent progress.
private void syncBackgroundWorker_ProgressChanged(object sender, ProgressChangedEventArgs e)
{
// Set the progress bar percent completed
pbStatus.Value = e.ProgressPercentage;
// Append the last agent message
tbLastStatusMessage.Text += e.UserState.ToString() + Environment.NewLine;
// Scroll to end
ScrollToEnd();
}
// This event handler deals with the results of the background operation.
private void syncBackgroundWorker_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e)
{
if (e.Cancelled == true)
{
tbLastStatusMessage.Text += "Canceled!" + Environment.NewLine;
ScrollToEnd();
}
else if (e.Error != null)
{
tbLastStatusMessage.Text += "Error: " + e.Error.Message + Environment.NewLine;
ScrollToEnd();
}
else
{
tbLastStatusMessage.Text += "Done!" + Environment.NewLine;
ScrollToEnd();
}
btnStart.Enabled = true;
pictureBoxStatus.Enabled = false;
}
private void ScrollToEnd()
{
// Scroll to end
tbLastStatusMessage.SelectionStart = tbLastStatusMessage.TextLength;
tbLastStatusMessage.ScrollToCaret();
}
private void btnClose_Click(object sender, EventArgs e)
{
this.Close();
}
}`